style(sync): expand Python and Kotlin workflows

This commit is contained in:
fy59 2026-09-17 14:45:00 +02:00
parent 6cd6219ac5
commit 7b769076c1
15 changed files with 2802 additions and 853 deletions

View file

@ -15,10 +15,15 @@ import org.json.JSONObject
internal sealed interface AndroidBackupResult {
data class Success(val bytes: Long) : AndroidBackupResult
data class Error(val message: String) : AndroidBackupResult
}
internal data class VerifiedAndroidBackup(val directory: File, val database: File, val preferences: File)
internal data class VerifiedAndroidBackup(
val directory: File,
val database: File,
val preferences: File,
)
/** Complete same-installation backup container; plaintext is explicit by design. */
internal class AndroidBackupService(private val context: Context) {
@ -34,26 +39,58 @@ internal class AndroidBackupService(private val context: Context) {
val value = MessageDigest.getInstance("SHA-256")
file.inputStream().use { input ->
val buffer = ByteArray(64 * 1024)
while (true) { val count = input.read(buffer); if (count < 0) break; value.update(buffer, 0, count) }
while (true) {
val count = input.read(buffer)
if (count < 0) break
value.update(buffer, 0, count)
}
}
return value.digest().joinToString("") { "%02x".format(it) }
}
private fun preferencesJson(): String = JSONObject().apply {
put("format", "trainlog-android-preferences"); put("version", 1)
put("stores", JSONObject().also { stores -> PREFERENCES.forEach { name ->
stores.put(name, JSONObject().also { values ->
context.getSharedPreferences(name, Context.MODE_PRIVATE).all.toSortedMap().forEach { (key, value) ->
when (value) {
null -> values.put(key, JSONObject.NULL)
is String, is Boolean, is Int, is Long, is Float -> values.put(key, value)
is Set<*> -> values.put(key, JSONArray(value.filterIsInstance<String>().sorted()))
else -> error("Unsupported preference type")
}
}
})
} })
}.toString()
private fun preferencesJson(): String =
JSONObject()
.apply {
put("format", "trainlog-android-preferences")
put("version", 1)
put(
"stores",
JSONObject().also { stores ->
PREFERENCES.forEach { name ->
stores.put(
name,
JSONObject().also { values ->
context
.getSharedPreferences(name, Context.MODE_PRIVATE)
.all
.toSortedMap()
.forEach { (key, value) ->
when (value) {
null -> values.put(key, JSONObject.NULL)
is String,
is Boolean,
is Int,
is Long,
is Float -> values.put(key, value)
is Set<*> ->
values.put(
key,
JSONArray(
value
.filterIsInstance<String>()
.sorted()
),
)
else -> error("Unsupported preference type")
}
}
},
)
}
},
)
}
.toString()
fun createPrivate(repository: TrainlogRepository): Result<File> = runCatching {
val directory = File(context.filesDir, "backups").apply { mkdirs() }
@ -66,85 +103,197 @@ internal class AndroidBackupService(private val context: Context) {
}
fun create(repository: TrainlogRepository, destination: OutputStream): AndroidBackupResult {
val archive = createPrivate(repository).getOrElse { return AndroidBackupResult.Error(it.message ?: "Backup failed.") }
return try { archive.inputStream().use { it.copyTo(destination) }; destination.flush()
val bytes = archive.length(); archive.delete(); AndroidBackupResult.Success(bytes)
val archive =
createPrivate(repository).getOrElse {
return AndroidBackupResult.Error(it.message ?: "Backup failed.")
}
return try {
archive.inputStream().use { it.copyTo(destination) }
destination.flush()
val bytes = archive.length()
archive.delete()
AndroidBackupResult.Success(bytes)
} catch (error: Exception) {
AndroidBackupResult.Error("Export failed; private backup retained at ${archive.name}: ${error.message}")
AndroidBackupResult.Error(
"Export failed; private backup retained at ${archive.name}: ${error.message}"
)
}
}
private fun writeArchive(repository: TrainlogRepository, destination: OutputStream): AndroidBackupResult {
private fun writeArchive(
repository: TrainlogRepository,
destination: OutputStream,
): AndroidBackupResult {
val stage = File(context.cacheDir, "backup-${java.util.UUID.randomUUID()}")
if (!stage.mkdir()) return AndroidBackupResult.Error("Cannot create private backup staging.")
val database = File(stage, "database.sqlite"); val preferences = File(stage, "preferences.json")
if (!stage.mkdir())
return AndroidBackupResult.Error("Cannot create private backup staging.")
val database = File(stage, "database.sqlite")
val preferences = File(stage, "preferences.json")
return try {
repository.backupDatabaseTo(database); preferences.writeText(preferencesJson())
repository.backupDatabaseTo(database)
preferences.writeText(preferencesJson())
val schema = validateDatabase(database)
val files = listOf("database.sqlite" to database, "preferences.json" to preferences)
val manifest = JSONObject().put("format", FORMAT).put("version", VERSION)
.put("package", context.packageName).put("schema", schema).put("product_version", "0.1.2")
.put("plaintext", true).put("created_at", OffsetDateTime.now().toString())
.put("entries", JSONArray(files.map { (name, file) -> JSONObject().put("path", name)
.put("size", file.length()).put("sha256", digest(file)) }))
val manifest =
JSONObject()
.put("format", FORMAT)
.put("version", VERSION)
.put("package", context.packageName)
.put("schema", schema)
.put("product_version", "0.1.2")
.put("plaintext", true)
.put("created_at", OffsetDateTime.now().toString())
.put(
"entries",
JSONArray(
files.map { (name, file) ->
JSONObject()
.put("path", name)
.put("size", file.length())
.put("sha256", digest(file))
}
),
)
var written = 0L
ZipOutputStream(destination).use { zip ->
files.forEach { (name, file) -> zip.putNextEntry(ZipEntry(name)); file.inputStream().use { input -> written += input.copyTo(zip) }; zip.closeEntry() }
zip.putNextEntry(ZipEntry("manifest.json")); val raw = manifest.toString().toByteArray(); zip.write(raw); written += raw.size; zip.closeEntry()
files.forEach { (name, file) ->
zip.putNextEntry(ZipEntry(name))
file.inputStream().use { input -> written += input.copyTo(zip) }
zip.closeEntry()
}
zip.putNextEntry(ZipEntry("manifest.json"))
val raw = manifest.toString().toByteArray()
zip.write(raw)
written += raw.size
zip.closeEntry()
}
AndroidBackupResult.Success(written)
} catch (error: Exception) { AndroidBackupResult.Error(error.message ?: "Backup failed.")
} finally { stage.deleteRecursively() }
} catch (error: Exception) {
AndroidBackupResult.Error(error.message ?: "Backup failed.")
} finally {
stage.deleteRecursively()
}
}
fun verify(input: InputStream): Result<VerifiedAndroidBackup> {
val stage = File(context.cacheDir, "restore-${java.util.UUID.randomUUID()}")
if (!stage.mkdir()) return Result.failure(IllegalStateException("Cannot create restore staging."))
if (!stage.mkdir())
return Result.failure(IllegalStateException("Cannot create restore staging."))
return runCatching {
var total = 0L; val names = mutableSetOf<String>()
ZipInputStream(input).use { zip ->
while (true) {
val entry = zip.nextEntry ?: break
require(!entry.isDirectory && entry.name in setOf("database.sqlite", "preferences.json", "manifest.json") && names.add(entry.name)) { "Invalid backup entry." }
val target = File(stage, entry.name); target.outputStream().use { output ->
val buffer = ByteArray(64 * 1024); var entryBytes = 0L
while (true) { val count = zip.read(buffer); if (count < 0) break; entryBytes += count; total += count
require(entryBytes <= MAX_ENTRY_BYTES && total <= MAX_TOTAL_BYTES) { "Backup exceeds bounds." }; output.write(buffer, 0, count) }
var total = 0L
val names = mutableSetOf<String>()
ZipInputStream(input).use { zip ->
while (true) {
val entry = zip.nextEntry ?: break
require(
!entry.isDirectory &&
entry.name in
setOf("database.sqlite", "preferences.json", "manifest.json") &&
names.add(entry.name)
) {
"Invalid backup entry."
}
val target = File(stage, entry.name)
target.outputStream().use { output ->
val buffer = ByteArray(64 * 1024)
var entryBytes = 0L
while (true) {
val count = zip.read(buffer)
if (count < 0) break
entryBytes += count
total += count
require(entryBytes <= MAX_ENTRY_BYTES && total <= MAX_TOTAL_BYTES) {
"Backup exceeds bounds."
}
output.write(buffer, 0, count)
}
}
}
}
require(names == setOf("database.sqlite", "preferences.json", "manifest.json")) {
"Backup entries are incomplete."
}
val manifest = JSONObject(File(stage, "manifest.json").readText())
require(
manifest.getString("format") == FORMAT &&
manifest.getInt("version") == VERSION &&
manifest.getString("package") == context.packageName &&
manifest.getBoolean("plaintext") &&
manifest.getInt("schema") in 17..20
) {
"Unsupported backup."
}
val expected = manifest.getJSONArray("entries")
require(expected.length() == 2)
val manifestPaths = mutableSetOf<String>()
for (index in 0 until expected.length()) {
val item = expected.getJSONObject(index)
val path = item.getString("path")
require(
path in setOf("database.sqlite", "preferences.json") &&
manifestPaths.add(path)
) {
"Invalid manifest entry."
}
val file = File(stage, path)
require(
file.canonicalFile.parentFile == stage.canonicalFile &&
file.length() == item.getLong("size") &&
digest(file) == item.getString("sha256")
) {
"Backup digest mismatch."
}
}
require(manifestPaths == setOf("database.sqlite", "preferences.json")) {
"Backup manifest is incomplete."
}
require(
validateDatabase(File(stage, "database.sqlite")) == manifest.getInt("schema")
)
val preferences = JSONObject(File(stage, "preferences.json").readText())
require(
preferences.getString("format") == "trainlog-android-preferences" &&
preferences.getInt("version") == 1
)
VerifiedAndroidBackup(
stage,
File(stage, "database.sqlite"),
File(stage, "preferences.json"),
)
}
}
require(names == setOf("database.sqlite", "preferences.json", "manifest.json")) { "Backup entries are incomplete." }
val manifest = JSONObject(File(stage, "manifest.json").readText())
require(manifest.getString("format") == FORMAT && manifest.getInt("version") == VERSION &&
manifest.getString("package") == context.packageName && manifest.getBoolean("plaintext") &&
manifest.getInt("schema") in 17..20) { "Unsupported backup." }
val expected = manifest.getJSONArray("entries")
require(expected.length() == 2)
val manifestPaths = mutableSetOf<String>()
for (index in 0 until expected.length()) { val item = expected.getJSONObject(index); val path = item.getString("path")
require(path in setOf("database.sqlite", "preferences.json") && manifestPaths.add(path)) { "Invalid manifest entry." }; val file = File(stage, path)
require(file.canonicalFile.parentFile == stage.canonicalFile && file.length() == item.getLong("size") && digest(file) == item.getString("sha256")) { "Backup digest mismatch." } }
require(manifestPaths == setOf("database.sqlite", "preferences.json")) { "Backup manifest is incomplete." }
require(validateDatabase(File(stage, "database.sqlite")) == manifest.getInt("schema"))
val preferences = JSONObject(File(stage, "preferences.json").readText())
require(preferences.getString("format") == "trainlog-android-preferences" && preferences.getInt("version") == 1)
VerifiedAndroidBackup(stage, File(stage, "database.sqlite"), File(stage, "preferences.json"))
}.onFailure { stage.deleteRecursively() }
.onFailure { stage.deleteRecursively() }
}
fun restore(repository: TrainlogRepository, verified: VerifiedAndroidBackup, databaseName: String = "trainlog-android.db"): AndroidBackupResult {
fun restore(
repository: TrainlogRepository,
verified: VerifiedAndroidBackup,
databaseName: String = "trainlog-android.db",
): AndroidBackupResult {
val destination = context.getDatabasePath(databaseName)
val previous = File(destination.parentFile, "${destination.name}.pre-restore-${java.util.UUID.randomUUID()}")
val previousPreferences = File(context.cacheDir, "preferences-pre-restore-${java.util.UUID.randomUUID()}.json")
val previous =
File(
destination.parentFile,
"${destination.name}.pre-restore-${java.util.UUID.randomUUID()}",
)
val previousPreferences =
File(context.cacheDir, "preferences-pre-restore-${java.util.UUID.randomUUID()}.json")
var moved = false
return try {
previousPreferences.writeText(preferencesJson())
repository.close(); destination.parentFile?.mkdirs()
if (destination.exists()) { check(destination.renameTo(previous)); moved = true }
File("${destination.path}-wal").delete(); File("${destination.path}-shm").delete()
repository.close()
destination.parentFile?.mkdirs()
if (destination.exists()) {
check(destination.renameTo(previous))
moved = true
}
File("${destination.path}-wal").delete()
File("${destination.path}-shm").delete()
verified.database.copyTo(destination, overwrite = false)
validateDatabase(destination); restorePreferences(verified.preferences)
previous.delete(); previousPreferences.delete(); verified.directory.deleteRecursively()
validateDatabase(destination)
restorePreferences(verified.preferences)
previous.delete()
previousPreferences.delete()
verified.directory.deleteRecursively()
AndroidBackupResult.Success(destination.length())
} catch (error: Exception) {
destination.delete()
@ -157,20 +306,40 @@ internal class AndroidBackupService(private val context: Context) {
private fun validateDatabase(file: File): Int {
val db = SQLiteDatabase.openDatabase(file.path, null, SQLiteDatabase.OPEN_READONLY)
try { require(db.rawQuery("PRAGMA integrity_check", null).use { it.moveToFirst() && it.getString(0) == "ok" })
try {
require(
db.rawQuery("PRAGMA integrity_check", null).use {
it.moveToFirst() && it.getString(0) == "ok"
}
)
require(db.rawQuery("PRAGMA foreign_key_check", null).use { !it.moveToFirst() })
require(db.version in 17..20) { "Unsupported Android schema ${db.version}." }
return db.version
} finally { db.close() }
} finally {
db.close()
}
}
private fun restorePreferences(file: File) {
val stores = JSONObject(file.readText()).getJSONObject("stores")
PREFERENCES.forEach { name -> val editor = context.getSharedPreferences(name, Context.MODE_PRIVATE).edit().clear(); val values = stores.getJSONObject(name)
values.keys().forEach { key -> when (val value = values.get(key)) {
is String -> editor.putString(key, value); is Boolean -> editor.putBoolean(key, value)
is Int -> editor.putInt(key, value); is Long -> editor.putLong(key, value)
is Double -> editor.putFloat(key, value.toFloat()); is JSONArray -> editor.putStringSet(key, (0 until value.length()).map { value.getString(it) }.toSet())
} }; check(editor.commit()) }
PREFERENCES.forEach { name ->
val editor = context.getSharedPreferences(name, Context.MODE_PRIVATE).edit().clear()
val values = stores.getJSONObject(name)
values.keys().forEach { key ->
when (val value = values.get(key)) {
is String -> editor.putString(key, value)
is Boolean -> editor.putBoolean(key, value)
is Int -> editor.putInt(key, value)
is Long -> editor.putLong(key, value)
is Double -> editor.putFloat(key, value.toFloat())
is JSONArray ->
editor.putStringSet(
key,
(0 until value.length()).map { value.getString(it) }.toSet(),
)
}
}
check(editor.commit())
}
}
}

View file

@ -9,6 +9,7 @@ import org.json.JSONObject
internal sealed interface ForegroundGenerationResult {
data class Completed(val runId: String) : ForegroundGenerationResult
data class Error(val message: String) : ForegroundGenerationResult
}
@ -19,43 +20,86 @@ internal class SyncGenerationForegroundCoordinator(private val repository: Train
private fun publish(path: File, value: String) {
val temporary = File(path.parentFile, ".${path.name}.tmp-${UUID.randomUUID()}")
FileOutputStream(temporary).use { stream ->
stream.write(value.toByteArray()); stream.flush(); stream.fd.sync()
stream.write(value.toByteArray())
stream.flush()
stream.fd.sync()
}
if (!temporary.renameTo(path)) throw SyncGenerationException("cannot publish ${path.name}")
}
private fun await(path: File, deadline: Long): File {
while (!path.isFile) {
if (System.nanoTime() >= deadline) throw SyncGenerationException("timeout waiting for ${path.name}")
if (System.nanoTime() >= deadline)
throw SyncGenerationException("timeout waiting for ${path.name}")
Thread.sleep(50)
}
return path
}
fun run(directory: File, timeout: Duration = Duration.ofMinutes(5), afterPublication: (() -> Unit)? = null): ForegroundGenerationResult = try {
if (!directory.isDirectory) throw SyncGenerationException("exchange directory is unavailable")
val peer = service.peerId()
publish(File(directory, "android-peer-v1.json"), JSONObject()
.put("format", "trainlog-sync-peer").put("version", 1).put("peer_id", peer)
.put("capabilities", JSONArray(listOf("generation-manifest-v1", "generation-ack-v1",
"causal-delete-v1", "mobile-history-v4", "execution-draft-v1"))).toString())
val deadline = System.nanoTime() + timeout.toNanos()
val request = JSONObject(await(File(directory, "request-v1.json"), deadline).readText())
if (request.getString("android_peer_id") != peer) throw SyncGenerationException("request targets another Android peer")
val captured = service.capture(directory, request.getString("desktop_peer_id"), request.getString("run_id"))
val published = service.publish(captured, File(directory, "android-objects"))
publish(File(directory, "android-generation-v1.json"), JSONObject()
.put("format", "trainlog-sync-generation-reference").put("version", 1)
.put("run_id", captured.runId).put("generation_id", captured.generationId)
.put("manifest_sha256", captured.manifestSha256)
.put("relative_path", published.relativeTo(directory).path).toString())
afterPublication?.invoke()
service.acceptAcknowledgement(await(File(directory, "desktop-consumption-ack-v1.json"), deadline).readBytes())
val reference = JSONObject(await(File(directory, "desktop-generation-v1.json"), deadline).readText())
val acknowledgement = service.consume(File(directory, reference.getString("relative_path")))
publish(File(directory, "android-consumption-ack-v1.json"), acknowledgement)
ForegroundGenerationResult.Completed(captured.runId)
} catch (error: Exception) {
ForegroundGenerationResult.Error(error.message ?: "generation synchronization failed")
}
fun run(
directory: File,
timeout: Duration = Duration.ofMinutes(5),
afterPublication: (() -> Unit)? = null,
): ForegroundGenerationResult =
try {
if (!directory.isDirectory)
throw SyncGenerationException("exchange directory is unavailable")
val peer = service.peerId()
publish(
File(directory, "android-peer-v1.json"),
JSONObject()
.put("format", "trainlog-sync-peer")
.put("version", 1)
.put("peer_id", peer)
.put(
"capabilities",
JSONArray(
listOf(
"generation-manifest-v1",
"generation-ack-v1",
"causal-delete-v1",
"mobile-history-v4",
"execution-draft-v1",
)
),
)
.toString(),
)
val deadline = System.nanoTime() + timeout.toNanos()
val request = JSONObject(await(File(directory, "request-v1.json"), deadline).readText())
if (request.getString("android_peer_id") != peer)
throw SyncGenerationException("request targets another Android peer")
val captured =
service.capture(
directory,
request.getString("desktop_peer_id"),
request.getString("run_id"),
)
val published = service.publish(captured, File(directory, "android-objects"))
publish(
File(directory, "android-generation-v1.json"),
JSONObject()
.put("format", "trainlog-sync-generation-reference")
.put("version", 1)
.put("run_id", captured.runId)
.put("generation_id", captured.generationId)
.put("manifest_sha256", captured.manifestSha256)
.put("relative_path", published.relativeTo(directory).path)
.toString(),
)
afterPublication?.invoke()
service.acceptAcknowledgement(
await(File(directory, "desktop-consumption-ack-v1.json"), deadline).readBytes()
)
val reference =
JSONObject(
await(File(directory, "desktop-generation-v1.json"), deadline).readText()
)
val acknowledgement =
service.consume(File(directory, reference.getString("relative_path")))
publish(File(directory, "android-consumption-ack-v1.json"), acknowledgement)
ForegroundGenerationResult.Completed(captured.runId)
} catch (error: Exception) {
ForegroundGenerationResult.Error(error.message ?: "generation synchronization failed")
}
}

View file

@ -5,8 +5,8 @@ import androidx.test.core.app.ApplicationProvider
import com.labfytools.trainlog.model.ActiveSessionDraft
import com.labfytools.trainlog.model.NewExerciseProfile
import com.labfytools.trainlog.model.RecordingMode
import com.labfytools.trainlog.model.SessionExerciseDraft
import com.labfytools.trainlog.model.SessionDraftForm
import com.labfytools.trainlog.model.SessionExerciseDraft
import com.labfytools.trainlog.model.SessionSetDraft
import com.labfytools.trainlog.model.TrackingMode
import java.io.ByteArrayInputStream
@ -24,20 +24,43 @@ import org.robolectric.annotation.Config
class AndroidBackupServiceTest {
private val context: Context = ApplicationProvider.getApplicationContext()
@Test fun completeBackupRestoresDraftPeerLedgersAndPreferences() {
@Test
fun completeBackupRestoresDraftPeerLedgersAndPreferences() {
val name = "backup-${UUID.randomUUID()}.db"
var repository = TrainlogRepository(context, name)
val service = AndroidBackupService(context)
try {
val exercise = (repository.createExercise(NewExerciseProfile("Backup exercise",
RecordingMode.SETS, TrackingMode.REPS, 0)) as CreateExerciseResult.Created).exercise
assertEquals(ActiveDraftMutationResult.Saved, repository.saveActiveSessionDraft(ActiveSessionDraft(
exercises = listOf(SessionExerciseDraft(exercise = exercise,
sets = listOf(SessionSetDraft(reps = 9, weightKg = 31.5)))),
form = SessionDraftForm(selectedExercise = exercise, weightText = "31,"))))
val exercise =
(repository.createExercise(
NewExerciseProfile(
"Backup exercise",
RecordingMode.SETS,
TrackingMode.REPS,
0,
)
) as CreateExerciseResult.Created)
.exercise
assertEquals(
ActiveDraftMutationResult.Saved,
repository.saveActiveSessionDraft(
ActiveSessionDraft(
exercises =
listOf(
SessionExerciseDraft(
exercise = exercise,
sets = listOf(SessionSetDraft(reps = 9, weightKg = 31.5)),
)
),
form = SessionDraftForm(selectedExercise = exercise, weightText = "31,"),
)
),
)
val peer = SyncGenerationService(repository).peerId()
context.getSharedPreferences("trainlog_presentation_settings", Context.MODE_PRIVATE)
.edit().putString("language", "en").commit()
context
.getSharedPreferences("trainlog_presentation_settings", Context.MODE_PRIVATE)
.edit()
.putString("language", "en")
.commit()
val output = ByteArrayOutputStream()
assertTrue(service.create(repository, output) is AndroidBackupResult.Success)
val verified = service.verify(ByteArrayInputStream(output.toByteArray())).getOrThrow()
@ -47,15 +70,26 @@ class AndroidBackupServiceTest {
assertTrue(draft is ActiveDraftLoadResult.Loaded)
assertEquals("31,", (draft as ActiveDraftLoadResult.Loaded).draft.form.weightText)
assertEquals(peer, SyncGenerationService(repository).peerId())
assertEquals("en", context.getSharedPreferences("trainlog_presentation_settings", Context.MODE_PRIVATE).getString("language", null))
} finally { repository.close(); context.deleteDatabase(name) }
assertEquals(
"en",
context
.getSharedPreferences("trainlog_presentation_settings", Context.MODE_PRIVATE)
.getString("language", null),
)
} finally {
repository.close()
context.deleteDatabase(name)
}
}
@Test fun corruptAndTraversalContainersAreRejectedBeforeRestore() {
@Test
fun corruptAndTraversalContainersAreRejectedBeforeRestore() {
val service = AndroidBackupService(context)
val output = ByteArrayOutputStream()
java.util.zip.ZipOutputStream(output).use { zip ->
zip.putNextEntry(java.util.zip.ZipEntry("../database.sqlite")); zip.write(byteArrayOf(1)); zip.closeEntry()
zip.putNextEntry(java.util.zip.ZipEntry("../database.sqlite"))
zip.write(byteArrayOf(1))
zip.closeEntry()
}
assertTrue(service.verify(ByteArrayInputStream(output.toByteArray())).isFailure)
}

View file

@ -2,8 +2,8 @@ package com.labfytools.trainlog.data
import android.content.Context
import androidx.test.core.app.ApplicationProvider
import java.io.FileInputStream
import java.io.File
import java.io.FileInputStream
import java.util.UUID
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
@ -13,19 +13,37 @@ import org.junit.runner.RunWith
import org.robolectric.RobolectricTestRunner
import org.robolectric.annotation.Config
@RunWith(RobolectricTestRunner::class) @Config(sdk=[35])
@RunWith(RobolectricTestRunner::class)
@Config(sdk = [35])
class BridgeBackupMigrationTest {
@Test fun verifiedSchemaSeventeenBridgeBackupMigratesThroughCurrentOwner() {
val source=System.getenv("TRAINLOG_BRIDGE_BACKUP_INPUT")?.let(::File)
assumeTrue(source?.isFile==true)
val context:Context=ApplicationProvider.getApplicationContext();val name="bridge-migrate-${UUID.randomUUID()}.db"
val service=AndroidBackupService(context);var repository=TrainlogRepository(context,name)
@Test
fun verifiedSchemaSeventeenBridgeBackupMigratesThroughCurrentOwner() {
val source = System.getenv("TRAINLOG_BRIDGE_BACKUP_INPUT")?.let(::File)
assumeTrue(source?.isFile == true)
val context: Context = ApplicationProvider.getApplicationContext()
val name = "bridge-migrate-${UUID.randomUUID()}.db"
val service = AndroidBackupService(context)
var repository = TrainlogRepository(context, name)
try {
val verified=FileInputStream(source!!).use { service.verify(it).getOrThrow() }
assertTrue(service.restore(repository,verified,name) is AndroidBackupResult.Success)
repository=TrainlogRepository(context,name)
assertEquals("21,",(repository.loadActiveSessionDraft() as ActiveDraftLoadResult.Loaded).draft.form.weightText)
android.database.sqlite.SQLiteDatabase.openDatabase(context.getDatabasePath(name).path,null,android.database.sqlite.SQLiteDatabase.OPEN_READONLY).use { assertEquals(20,it.version) }
} finally {repository.close();context.deleteDatabase(name)}
val verified = FileInputStream(source!!).use { service.verify(it).getOrThrow() }
assertTrue(service.restore(repository, verified, name) is AndroidBackupResult.Success)
repository = TrainlogRepository(context, name)
assertEquals(
"21,",
(repository.loadActiveSessionDraft() as ActiveDraftLoadResult.Loaded)
.draft
.form
.weightText,
)
android.database.sqlite.SQLiteDatabase.openDatabase(
context.getDatabasePath(name).path,
null,
android.database.sqlite.SQLiteDatabase.OPEN_READONLY,
)
.use { assertEquals(20, it.version) }
} finally {
repository.close()
context.deleteDatabase(name)
}
}
}

View file

@ -36,7 +36,9 @@ class SyncDeploymentConversationTest {
private fun publish(file: File, value: String) {
val temporary = File(file.parentFile, ".${file.name}.tmp-${UUID.randomUUID()}")
temporary.outputStream().use { output ->
output.write(value.toByteArray()); output.flush(); output.fd.sync()
output.write(value.toByteArray())
output.flush()
output.fd.sync()
}
check(temporary.renameTo(file)) { "cannot publish ${file.name}" }
}
@ -44,42 +46,83 @@ class SyncDeploymentConversationTest {
@Test
fun foregroundAndroidServiceCompletesCorrelatedDesktopConversation() {
val suppliedTransport = System.getenv("TRAINLOG_SYNC_TRANSPORT_ROOT")
val transport = suppliedTransport?.let(::File) ?: kotlin.io.path.createTempDirectory("deployment-conversation-").toFile()
val repositoryRoot = generateSequence(File(checkNotNull(System.getProperty("user.dir")))) { it.parentFile }
.first { File(it, "tools/sync_peer_worker.py").isFile }
val transport =
suppliedTransport?.let(::File)
?: kotlin.io.path.createTempDirectory("deployment-conversation-").toFile()
val repositoryRoot =
generateSequence(File(checkNotNull(System.getProperty("user.dir")))) { it.parentFile }
.first { File(it, "tools/sync_peer_worker.py").isFile }
val databaseName = "deployment-peer-${UUID.randomUUID()}.db"
var repository = TrainlogRepository(context, databaseName)
var localWorker: Process? = null
try {
transport.mkdirs()
val exercise = (repository.createExercise(NewExerciseProfile(
"Deployment conversation", RecordingMode.SETS, TrackingMode.REPS, 0,
)) as CreateExerciseResult.Created).exercise
val occurrence = SessionExerciseDraft(exercise = exercise,
sets = listOf(SessionSetDraft(reps = 7, weightKg = 22.5)))
assertEquals(ActiveDraftMutationResult.Saved,
repository.saveActiveSessionDraft(ActiveSessionDraft(exercises = listOf(occurrence))))
val exercise =
(repository.createExercise(
NewExerciseProfile(
"Deployment conversation",
RecordingMode.SETS,
TrackingMode.REPS,
0,
)
) as CreateExerciseResult.Created)
.exercise
val occurrence =
SessionExerciseDraft(
exercise = exercise,
sets = listOf(SessionSetDraft(reps = 7, weightKg = 22.5)),
)
assertEquals(
ActiveDraftMutationResult.Saved,
repository.saveActiveSessionDraft(
ActiveSessionDraft(exercises = listOf(occurrence))
),
)
val coordinator = SyncGenerationForegroundCoordinator(repository)
val release = System.getenv("TRAINLOG_SYNC_TEST_RELEASE_FILE")
val executor = java.util.concurrent.Executors.newSingleThreadExecutor()
val result = executor.submit<ForegroundGenerationResult> {
coordinator.run(transport, Duration.ofSeconds(30)) {
release?.let { awaitFile(File(it)) }
val result =
executor.submit<ForegroundGenerationResult> {
coordinator.run(transport, Duration.ofSeconds(30)) {
release?.let { awaitFile(File(it)) }
}
}
}
val peer = JSONObject(awaitFile(File(transport, "android-peer-v1.json")).readText())
if (suppliedTransport == null) {
val desktop = File(transport, "desktop.sqlite")
val fixture = ProcessBuilder(File(repositoryRoot, "build/tui/sync-generation-fixture").absolutePath,
desktop.absolutePath).directory(repositoryRoot).start()
val fixture =
ProcessBuilder(
File(repositoryRoot, "build/tui/sync-generation-fixture").absolutePath,
desktop.absolutePath,
)
.directory(repositoryRoot)
.start()
assertEquals(0, fixture.waitFor())
localWorker = ProcessBuilder("python3", File(repositoryRoot, "tools/sync_peer_worker.py").absolutePath,
"--database", desktop.absolutePath, "--transport-root", transport.absolutePath,
"--owned-root", File(transport, "owned").absolutePath, "--run-id", "sy_${UUID.randomUUID()}",
"--expected-peer", peer.getString("peer_id"), "--timeout", "30").directory(repositoryRoot)
.redirectErrorStream(true).start()
localWorker =
ProcessBuilder(
"python3",
File(repositoryRoot, "tools/sync_peer_worker.py").absolutePath,
"--database",
desktop.absolutePath,
"--transport-root",
transport.absolutePath,
"--owned-root",
File(transport, "owned").absolutePath,
"--run-id",
"sy_${UUID.randomUUID()}",
"--expected-peer",
peer.getString("peer_id"),
"--timeout",
"30",
)
.directory(repositoryRoot)
.redirectErrorStream(true)
.start()
}
assertTrue(result.get(35, java.util.concurrent.TimeUnit.SECONDS) is ForegroundGenerationResult.Completed)
assertTrue(
result.get(35, java.util.concurrent.TimeUnit.SECONDS)
is ForegroundGenerationResult.Completed
)
executor.shutdownNow()
repository.close()
repository = TrainlogRepository(context, databaseName)

View file

@ -9,6 +9,10 @@ import com.labfytools.trainlog.model.SessionDraft
import com.labfytools.trainlog.model.SessionExerciseDraft
import com.labfytools.trainlog.model.SessionSetDraft
import com.labfytools.trainlog.model.TrackingMode
import java.nio.file.Files
import java.util.UUID
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import org.json.JSONObject
import org.junit.Assert.assertEquals
import org.junit.Assert.assertTrue
@ -16,10 +20,6 @@ import org.junit.Test
import org.junit.runner.RunWith
import org.robolectric.RobolectricTestRunner
import org.robolectric.annotation.Config
import java.nio.file.Files
import java.util.UUID
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
@RunWith(RobolectricTestRunner::class)
@Config(sdk = [35])
@ -28,157 +28,493 @@ class SyncGenerationServiceTest {
@Test
fun v19MigrationPreservesDataAndInventsNoGenerationState() {
val name="generation-migration-${UUID.randomUUID()}.db"
TrainlogRepository(context,name).also{
assertTrue(it.createExercise(NewExerciseProfile("Migration sentinel",RecordingMode.SETS,TrackingMode.REPS,0)) is CreateExerciseResult.Created)
val name = "generation-migration-${UUID.randomUUID()}.db"
TrainlogRepository(context, name).also {
assertTrue(
it.createExercise(
NewExerciseProfile(
"Migration sentinel",
RecordingMode.SETS,
TrackingMode.REPS,
0,
)
) is CreateExerciseResult.Created
)
it.close()
}
val path=context.getDatabasePath(name)
SQLiteDatabase.openDatabase(path.absolutePath,null,SQLiteDatabase.OPEN_READWRITE).use{db->
listOf("sync_causal_publications","sync_acknowledgements","sync_consumed_generations","sync_generation_artifacts","sync_generations","sync_peer_identity").forEach{db.execSQL("DROP TABLE $it")}
val path = context.getDatabasePath(name)
SQLiteDatabase.openDatabase(path.absolutePath, null, SQLiteDatabase.OPEN_READWRITE).use { db
->
listOf(
"sync_causal_publications",
"sync_acknowledgements",
"sync_consumed_generations",
"sync_generation_artifacts",
"sync_generations",
"sync_peer_identity",
)
.forEach { db.execSQL("DROP TABLE $it") }
db.execSQL("PRAGMA user_version=19")
}
val migrated=TrainlogRepository(context,name).also{it.listExercises()}
try{
SQLiteDatabase.openDatabase(path.absolutePath,null,SQLiteDatabase.OPEN_READONLY).use{db->
db.rawQuery("PRAGMA user_version",null).use{assertTrue(it.moveToFirst());assertEquals(20,it.getInt(0))}
db.rawQuery("SELECT COUNT(*) FROM exercises",null).use{assertTrue(it.moveToFirst());assertTrue(it.getInt(0)>0)}
listOf("sync_generations","sync_consumed_generations","sync_acknowledgements","sync_causal_publications").forEach{table->db.rawQuery("SELECT COUNT(*) FROM $table",null).use{assertTrue(it.moveToFirst());assertEquals(0,it.getInt(0))}}
}
}finally{migrated.close();context.deleteDatabase(name)}
val migrated = TrainlogRepository(context, name).also { it.listExercises() }
try {
SQLiteDatabase.openDatabase(path.absolutePath, null, SQLiteDatabase.OPEN_READONLY)
.use { db ->
db.rawQuery("PRAGMA user_version", null).use {
assertTrue(it.moveToFirst())
assertEquals(20, it.getInt(0))
}
db.rawQuery("SELECT COUNT(*) FROM exercises", null).use {
assertTrue(it.moveToFirst())
assertTrue(it.getInt(0) > 0)
}
listOf(
"sync_generations",
"sync_consumed_generations",
"sync_acknowledgements",
"sync_causal_publications",
)
.forEach { table ->
db.rawQuery("SELECT COUNT(*) FROM $table", null).use {
assertTrue(it.moveToFirst())
assertEquals(0, it.getInt(0))
}
}
}
} finally {
migrated.close()
context.deleteDatabase(name)
}
}
@Test
fun lateSemanticFailureRollsBackWholeAndroidGeneration() {
val sourceName="generation-rollback-source-${UUID.randomUUID()}.db";val destinationName="generation-rollback-destination-${UUID.randomUUID()}.db"
val root=Files.createTempDirectory("trainlog-generation-rollback-").toFile();val source=TrainlogRepository(context,sourceName);val destination=TrainlogRepository(context,destinationName)
try{
val exercise=(source.createExercise(NewExerciseProfile("Rollback exercise",RecordingMode.SETS,TrackingMode.REPS,0)) as CreateExerciseResult.Created).exercise
assertTrue(source.saveSession(SessionDraft(listOf(SessionExerciseDraft(exercise=exercise,sets=listOf(SessionSetDraft(reps=5)))))) is SaveSessionResult.Saved)
val destinationService=SyncGenerationService(destination);val identity=destinationService.capture(root,"peer_11111111-1111-4111-8111-111111111111")
val peer=JSONObject(java.io.File(identity.stagingDirectory,"manifest.json").readText()).getJSONObject("producer").getString("peer_id")
val sourceService=SyncGenerationService(source);val captured=sourceService.capture(root,peer);val published=sourceService.publish(captured,java.io.File(root,"objects"))
val bad="{}".toByteArray();java.io.File(published,"feedback-v2.json").writeBytes(bad)
val manifest=JSONObject(java.io.File(published,"manifest.json").readText());val artifacts=manifest.getJSONArray("artifacts")
for(index in 0 until artifacts.length())if(artifacts.getJSONObject(index).getString("logical_name")=="feedback")artifacts.getJSONObject(index).put("size",bad.size).put("sha256",java.security.MessageDigest.getInstance("SHA-256").digest(bad).joinToString(""){"%02x".format(it)})
java.io.File(published,"manifest.json").writeText(manifest.toString())
assertEquals("rejected",JSONObject(destinationService.consume(published)).getString("result"))
val sourceName = "generation-rollback-source-${UUID.randomUUID()}.db"
val destinationName = "generation-rollback-destination-${UUID.randomUUID()}.db"
val root = Files.createTempDirectory("trainlog-generation-rollback-").toFile()
val source = TrainlogRepository(context, sourceName)
val destination = TrainlogRepository(context, destinationName)
try {
val exercise =
(source.createExercise(
NewExerciseProfile(
"Rollback exercise",
RecordingMode.SETS,
TrackingMode.REPS,
0,
)
) as CreateExerciseResult.Created)
.exercise
assertTrue(
source.saveSession(
SessionDraft(
listOf(
SessionExerciseDraft(
exercise = exercise,
sets = listOf(SessionSetDraft(reps = 5)),
)
)
)
) is SaveSessionResult.Saved
)
val destinationService = SyncGenerationService(destination)
val identity =
destinationService.capture(root, "peer_11111111-1111-4111-8111-111111111111")
val peer =
JSONObject(java.io.File(identity.stagingDirectory, "manifest.json").readText())
.getJSONObject("producer")
.getString("peer_id")
val sourceService = SyncGenerationService(source)
val captured = sourceService.capture(root, peer)
val published = sourceService.publish(captured, java.io.File(root, "objects"))
val bad = "{}".toByteArray()
java.io.File(published, "feedback-v2.json").writeBytes(bad)
val manifest = JSONObject(java.io.File(published, "manifest.json").readText())
val artifacts = manifest.getJSONArray("artifacts")
for (index in 0 until artifacts.length()) if (
artifacts.getJSONObject(index).getString("logical_name") == "feedback"
)
artifacts
.getJSONObject(index)
.put("size", bad.size)
.put(
"sha256",
java.security.MessageDigest.getInstance("SHA-256").digest(bad).joinToString(
""
) {
"%02x".format(it)
},
)
java.io.File(published, "manifest.json").writeText(manifest.toString())
assertEquals(
"rejected",
JSONObject(destinationService.consume(published)).getString("result"),
)
assertTrue(destination.listSessions().isEmpty())
SQLiteDatabase.openDatabase(context.getDatabasePath(destinationName).absolutePath,null,SQLiteDatabase.OPEN_READONLY).use{db->db.rawQuery("SELECT COUNT(*) FROM sync_consumed_generations WHERE result='rejected'",null).use{assertTrue(it.moveToFirst());assertEquals(1,it.getInt(0))}}
}finally{source.close();destination.close();context.deleteDatabase(sourceName);context.deleteDatabase(destinationName);root.deleteRecursively()}
SQLiteDatabase.openDatabase(
context.getDatabasePath(destinationName).absolutePath,
null,
SQLiteDatabase.OPEN_READONLY,
)
.use { db ->
db.rawQuery(
"SELECT COUNT(*) FROM sync_consumed_generations WHERE result='rejected'",
null,
)
.use {
assertTrue(it.moveToFirst())
assertEquals(1, it.getInt(0))
}
}
} finally {
source.close()
destination.close()
context.deleteDatabase(sourceName)
context.deleteDatabase(destinationName)
root.deleteRecursively()
}
}
@Test
fun causalOperationBytesRemainImmutableAcrossAndroidGenerations() {
val name="generation-causal-${UUID.randomUUID()}.db";val root=Files.createTempDirectory("trainlog-generation-causal-").toFile();val repo=TrainlogRepository(context,name)
try{
val exercise=(repo.createExercise(NewExerciseProfile("Causal generation",RecordingMode.SETS,TrackingMode.REPS,0)) as CreateExerciseResult.Created).exercise
assertTrue(repo.saveSession(SessionDraft(listOf(SessionExerciseDraft(exercise=exercise,sets=listOf(SessionSetDraft(reps=4)))))) is SaveSessionResult.Saved)
val sessionId=repo.listSessions().single().sessionId
assertTrue(repo.deleteCausally("session",sessionId,"peer_android_generation") is CausalDeleteResult.Applied)
val service=SyncGenerationService(repo);val peer="peer_33333333-3333-4333-8333-333333333333"
val first=service.capture(root,peer);val second=service.capture(root,peer)
fun operation(captured:CapturedSyncGeneration)=JSONObject(java.io.File(captured.stagingDirectory,"causal-deletions-v1.json").readText()).getJSONArray("operations").getJSONObject(0)
assertEquals(operation(first).toString(),operation(second).toString());assertTrue(operation(first).isNull("publication_context"))
SQLiteDatabase.openDatabase(context.getDatabasePath(name).absolutePath,null,SQLiteDatabase.OPEN_READONLY).use{db->
db.rawQuery("SELECT SUM(first_emission),COUNT(*)-SUM(first_emission) FROM sync_causal_publications",null).use{assertTrue(it.moveToFirst());assertEquals(1,it.getInt(0));assertEquals(1,it.getInt(1))}
}
}finally{repo.close();context.deleteDatabase(name);root.deleteRecursively()}
val name = "generation-causal-${UUID.randomUUID()}.db"
val root = Files.createTempDirectory("trainlog-generation-causal-").toFile()
val repo = TrainlogRepository(context, name)
try {
val exercise =
(repo.createExercise(
NewExerciseProfile(
"Causal generation",
RecordingMode.SETS,
TrackingMode.REPS,
0,
)
) as CreateExerciseResult.Created)
.exercise
assertTrue(
repo.saveSession(
SessionDraft(
listOf(
SessionExerciseDraft(
exercise = exercise,
sets = listOf(SessionSetDraft(reps = 4)),
)
)
)
) is SaveSessionResult.Saved
)
val sessionId = repo.listSessions().single().sessionId
assertTrue(
repo.deleteCausally("session", sessionId, "peer_android_generation")
is CausalDeleteResult.Applied
)
val service = SyncGenerationService(repo)
val peer = "peer_33333333-3333-4333-8333-333333333333"
val first = service.capture(root, peer)
val second = service.capture(root, peer)
fun operation(captured: CapturedSyncGeneration) =
JSONObject(
java.io
.File(captured.stagingDirectory, "causal-deletions-v1.json")
.readText()
)
.getJSONArray("operations")
.getJSONObject(0)
assertEquals(operation(first).toString(), operation(second).toString())
assertTrue(operation(first).isNull("publication_context"))
SQLiteDatabase.openDatabase(
context.getDatabasePath(name).absolutePath,
null,
SQLiteDatabase.OPEN_READONLY,
)
.use { db ->
db.rawQuery(
"SELECT SUM(first_emission),COUNT(*)-SUM(first_emission) FROM sync_causal_publications",
null,
)
.use {
assertTrue(it.moveToFirst())
assertEquals(1, it.getInt(0))
assertEquals(1, it.getInt(1))
}
}
} finally {
repo.close()
context.deleteDatabase(name)
root.deleteRecursively()
}
}
@Test
fun androidCaptureUsesOneSnapshotAgainstControlledWriter() {
val name="generation-snapshot-${UUID.randomUUID()}.db";val repo=TrainlogRepository(context,name)
try{
val first=(repo.createExercise(NewExerciseProfile("Snapshot A",RecordingMode.SETS,TrackingMode.REPS,0)) as CreateExerciseResult.Created).exercise
assertTrue(repo.saveSession(SessionDraft(listOf(SessionExerciseDraft(exercise=first,sets=listOf(SessionSetDraft(reps=3)))))) is SaveSessionResult.Saved)
val attempted=CountDownLatch(1);var writer:Thread?=null
val captured=repo.captureSyncGenerationArtifacts{
writer=Thread{
attempted.countDown()
repo.createExercise(NewExerciseProfile("Snapshot B",RecordingMode.SETS,TrackingMode.REPS,0))
}.also{it.start()}
assertTrue(attempted.await(5,TimeUnit.SECONDS))
}
val completedWriter=checkNotNull(writer)
completedWriter.join(5000);assertTrue(!completedWriter.isAlive)
assertEquals(1,JSONObject(captured.getValue("catalog")).getJSONArray("exercises").length())
assertEquals(1,JSONObject(captured.getValue("exercise-profile-state")).getJSONArray("exercises").length())
assertEquals(2,repo.listExercises().size)
}finally{repo.close();context.deleteDatabase(name)}
val name = "generation-snapshot-${UUID.randomUUID()}.db"
val repo = TrainlogRepository(context, name)
try {
val first =
(repo.createExercise(
NewExerciseProfile("Snapshot A", RecordingMode.SETS, TrackingMode.REPS, 0)
) as CreateExerciseResult.Created)
.exercise
assertTrue(
repo.saveSession(
SessionDraft(
listOf(
SessionExerciseDraft(
exercise = first,
sets = listOf(SessionSetDraft(reps = 3)),
)
)
)
) is SaveSessionResult.Saved
)
val attempted = CountDownLatch(1)
var writer: Thread? = null
val captured =
repo.captureSyncGenerationArtifacts {
writer =
Thread {
attempted.countDown()
repo.createExercise(
NewExerciseProfile(
"Snapshot B",
RecordingMode.SETS,
TrackingMode.REPS,
0,
)
)
}
.also { it.start() }
assertTrue(attempted.await(5, TimeUnit.SECONDS))
}
val completedWriter = checkNotNull(writer)
completedWriter.join(5000)
assertTrue(!completedWriter.isAlive)
assertEquals(
1,
JSONObject(captured.getValue("catalog")).getJSONArray("exercises").length(),
)
assertEquals(
1,
JSONObject(captured.getValue("exercise-profile-state"))
.getJSONArray("exercises")
.length(),
)
assertEquals(2, repo.listExercises().size)
} finally {
repo.close()
context.deleteDatabase(name)
}
}
@Test
fun androidGenerationCapturePublishConsumeAndAckSurviveReopen() {
val sourceName="generation-source-${UUID.randomUUID()}.db"
val destinationName="generation-destination-${UUID.randomUUID()}.db"
val desktopDestinationName="generation-desktop-destination-${UUID.randomUUID()}.db"
val root=Files.createTempDirectory("trainlog-generation-android-").toFile()
var source=TrainlogRepository(context,sourceName);var destination=TrainlogRepository(context,destinationName)
var desktopDestination=TrainlogRepository(context,desktopDestinationName)
val sourceName = "generation-source-${UUID.randomUUID()}.db"
val destinationName = "generation-destination-${UUID.randomUUID()}.db"
val desktopDestinationName = "generation-desktop-destination-${UUID.randomUUID()}.db"
val root = Files.createTempDirectory("trainlog-generation-android-").toFile()
var source = TrainlogRepository(context, sourceName)
var destination = TrainlogRepository(context, destinationName)
var desktopDestination = TrainlogRepository(context, desktopDestinationName)
try {
val created=source.createExercise(NewExerciseProfile("Generation exercise",RecordingMode.SETS,TrackingMode.REPS,0)) as CreateExerciseResult.Created
val saved = source.saveSession(
SessionDraft(listOf(SessionExerciseDraft(
exercise=created.exercise,
sets=listOf(SessionSetDraft(reps=7,weightKg=21.5)),
))),
)
val created =
source.createExercise(
NewExerciseProfile(
"Generation exercise",
RecordingMode.SETS,
TrackingMode.REPS,
0,
)
) as CreateExerciseResult.Created
val saved =
source.saveSession(
SessionDraft(
listOf(
SessionExerciseDraft(
exercise = created.exercise,
sets = listOf(SessionSetDraft(reps = 7, weightKg = 21.5)),
)
)
)
)
assertTrue(saved is SaveSessionResult.Saved)
val destinationService=SyncGenerationService(destination)
val identityCapture=destinationService.capture(root,"peer_11111111-1111-4111-8111-111111111111")
val destinationPeer=JSONObject(java.io.File(identityCapture.stagingDirectory,"manifest.json").readText()).getJSONObject("producer").getString("peer_id")
val sourceService=SyncGenerationService(source)
val captured=sourceService.capture(root,destinationPeer)
val published=sourceService.publish(captured,java.io.File(root,"objects"))
val before=published.listFiles()!!.associate{it.name to it.readBytes().contentHashCode()}
assertEquals(published,sourceService.publish(captured,java.io.File(root,"objects")))
assertEquals(before,published.listFiles()!!.associate{it.name to it.readBytes().contentHashCode()})
val ack=destinationService.consume(published)
assertEquals("acknowledged",sourceService.acceptAcknowledgement(ack.toByteArray()))
source.close();destination.close()
source=TrainlogRepository(context,sourceName);destination=TrainlogRepository(context,destinationName)
assertEquals(1,destination.listSessions().size)
assertEquals("unchanged",SyncGenerationService(source).acceptAcknowledgement(ack.toByteArray()))
assertEquals(ack,SyncGenerationService(destination).consume(published))
val destinationService = SyncGenerationService(destination)
val identityCapture =
destinationService.capture(root, "peer_11111111-1111-4111-8111-111111111111")
val destinationPeer =
JSONObject(
java.io.File(identityCapture.stagingDirectory, "manifest.json").readText()
)
.getJSONObject("producer")
.getString("peer_id")
val sourceService = SyncGenerationService(source)
val captured = sourceService.capture(root, destinationPeer)
val published = sourceService.publish(captured, java.io.File(root, "objects"))
val before =
published.listFiles()!!.associate { it.name to it.readBytes().contentHashCode() }
assertEquals(published, sourceService.publish(captured, java.io.File(root, "objects")))
assertEquals(
before,
published.listFiles()!!.associate { it.name to it.readBytes().contentHashCode() },
)
val ack = destinationService.consume(published)
assertEquals("acknowledged", sourceService.acceptAcknowledgement(ack.toByteArray()))
source.close()
destination.close()
source = TrainlogRepository(context, sourceName)
destination = TrainlogRepository(context, destinationName)
assertEquals(1, destination.listSessions().size)
assertEquals(
"unchanged",
SyncGenerationService(source).acceptAcknowledgement(ack.toByteArray()),
)
assertEquals(ack, SyncGenerationService(destination).consume(published))
// A desktop-created mutation uses the public C persistence API,
// then the production desktop generation entry point. This is not
// an Android generation re-exported through desktop.
val repositoryRoot=generateSequence(java.io.File(checkNotNull(System.getProperty("user.dir")))){it.parentFile}
.first{java.io.File(it,"tools/sync_generation_exchange.py").isFile}
val desktopDb=java.io.File(root,"desktop-source.sqlite")
fun run(vararg command:String):String{
val process=ProcessBuilder(*command).directory(repositoryRoot).redirectErrorStream(true).start()
val text=process.inputStream.bufferedReader().readText()
assertEquals("command failed: ${command.joinToString(" ")}\n$text",0,process.waitFor())
val repositoryRoot =
generateSequence(java.io.File(checkNotNull(System.getProperty("user.dir")))) {
it.parentFile
}
.first { java.io.File(it, "tools/sync_generation_exchange.py").isFile }
val desktopDb = java.io.File(root, "desktop-source.sqlite")
fun run(vararg command: String): String {
val process =
ProcessBuilder(*command)
.directory(repositoryRoot)
.redirectErrorStream(true)
.start()
val text = process.inputStream.bufferedReader().readText()
assertEquals(
"command failed: ${command.joinToString(" ")}\n$text",
0,
process.waitFor(),
)
return text
}
val desktopConsumerDb=java.io.File(root,"desktop-consumer.sqlite")
run(java.io.File(repositoryRoot,"build/tui/sync-generation-fixture").absolutePath,desktopConsumerDb.absolutePath)
val desktopPeer=Regex("PEER_ID=(peer_[^\\s]+)").find(run("python3",java.io.File(repositoryRoot,"tools/sync_generation_exchange.py").absolutePath,"peer-id","--database",desktopConsumerDb.absolutePath))!!.groupValues[1]
val androidToDesktop=sourceService.capture(root,desktopPeer)
val androidPublished=sourceService.publish(androidToDesktop,java.io.File(root,"android-desktop-objects"))
val androidAckFile=java.io.File(root,"android-desktop-ack.json")
run("python3",java.io.File(repositoryRoot,"tools/sync_generation_exchange.py").absolutePath,"consume-desktop",androidPublished.absolutePath,"--database",desktopConsumerDb.absolutePath,"--ack-output",androidAckFile.absolutePath)
assertEquals("acknowledged",sourceService.acceptAcknowledgement(androidAckFile.readBytes()))
val androidSessionId=source.listSessions().single().sessionId
assertEquals("1",run("sqlite3",desktopConsumerDb.absolutePath,"SELECT count(*) FROM sessions WHERE session_id='$androidSessionId';").trim())
val desktopConsumerDb = java.io.File(root, "desktop-consumer.sqlite")
run(
java.io.File(repositoryRoot, "build/tui/sync-generation-fixture").absolutePath,
desktopConsumerDb.absolutePath,
)
val desktopPeer =
Regex("PEER_ID=(peer_[^\\s]+)")
.find(
run(
"python3",
java.io
.File(repositoryRoot, "tools/sync_generation_exchange.py")
.absolutePath,
"peer-id",
"--database",
desktopConsumerDb.absolutePath,
)
)!!
.groupValues[1]
val androidToDesktop = sourceService.capture(root, desktopPeer)
val androidPublished =
sourceService.publish(
androidToDesktop,
java.io.File(root, "android-desktop-objects"),
)
val androidAckFile = java.io.File(root, "android-desktop-ack.json")
run(
"python3",
java.io.File(repositoryRoot, "tools/sync_generation_exchange.py").absolutePath,
"consume-desktop",
androidPublished.absolutePath,
"--database",
desktopConsumerDb.absolutePath,
"--ack-output",
androidAckFile.absolutePath,
)
assertEquals(
"acknowledged",
sourceService.acceptAcknowledgement(androidAckFile.readBytes()),
)
val androidSessionId = source.listSessions().single().sessionId
assertEquals(
"1",
run(
"sqlite3",
desktopConsumerDb.absolutePath,
"SELECT count(*) FROM sessions WHERE session_id='$androidSessionId';",
)
.trim(),
)
run(java.io.File(repositoryRoot,"build/tui/sync-generation-fixture").absolutePath,desktopDb.absolutePath)
val androidPeerCapture=SyncGenerationService(desktopDestination).capture(root,"peer_22222222-2222-4222-8222-222222222222")
val androidPeer=JSONObject(java.io.File(androidPeerCapture.stagingDirectory,"manifest.json").readText()).getJSONObject("producer").getString("peer_id")
val desktopOwned=java.io.File(root,"desktop-owned");val desktopObjects=java.io.File(root,"desktop-objects")
val captureText=run("python3",java.io.File(repositoryRoot,"tools/sync_generation_exchange.py").absolutePath,"capture-desktop","--database",desktopDb.absolutePath,"--owned-root",desktopOwned.absolutePath,"--consumer-peer",androidPeer)
val desktopGeneration=Regex("generation_id=(gen_[^ ]+)").find(captureText)!!.groupValues[1]
run("python3",java.io.File(repositoryRoot,"tools/sync_generation_exchange.py").absolutePath,"publish","--database",desktopDb.absolutePath,"--generation-id",desktopGeneration,"--object-root",desktopObjects.absolutePath)
val desktopPublished=java.io.File(desktopObjects,"generations/$desktopGeneration")
val desktopAck=SyncGenerationService(desktopDestination).consume(desktopPublished)
val desktopAckFile=java.io.File(root,"desktop-ack.json").apply{writeText(desktopAck)}
assertTrue(run("python3",java.io.File(repositoryRoot,"tools/sync_generation_exchange.py").absolutePath,"accept-ack",desktopAckFile.absolutePath,"--database",desktopDb.absolutePath).contains("ACK_ACCEPT=acknowledged"))
assertTrue(desktopDestination.listSessions().any{it.sessionId=="se_99999999-9999-4999-8999-999999999999"})
run(
java.io.File(repositoryRoot, "build/tui/sync-generation-fixture").absolutePath,
desktopDb.absolutePath,
)
val androidPeerCapture =
SyncGenerationService(desktopDestination)
.capture(root, "peer_22222222-2222-4222-8222-222222222222")
val androidPeer =
JSONObject(
java.io
.File(androidPeerCapture.stagingDirectory, "manifest.json")
.readText()
)
.getJSONObject("producer")
.getString("peer_id")
val desktopOwned = java.io.File(root, "desktop-owned")
val desktopObjects = java.io.File(root, "desktop-objects")
val captureText =
run(
"python3",
java.io.File(repositoryRoot, "tools/sync_generation_exchange.py").absolutePath,
"capture-desktop",
"--database",
desktopDb.absolutePath,
"--owned-root",
desktopOwned.absolutePath,
"--consumer-peer",
androidPeer,
)
val desktopGeneration =
Regex("generation_id=(gen_[^ ]+)").find(captureText)!!.groupValues[1]
run(
"python3",
java.io.File(repositoryRoot, "tools/sync_generation_exchange.py").absolutePath,
"publish",
"--database",
desktopDb.absolutePath,
"--generation-id",
desktopGeneration,
"--object-root",
desktopObjects.absolutePath,
)
val desktopPublished = java.io.File(desktopObjects, "generations/$desktopGeneration")
val desktopAck = SyncGenerationService(desktopDestination).consume(desktopPublished)
val desktopAckFile =
java.io.File(root, "desktop-ack.json").apply { writeText(desktopAck) }
assertTrue(
run(
"python3",
java.io
.File(repositoryRoot, "tools/sync_generation_exchange.py")
.absolutePath,
"accept-ack",
desktopAckFile.absolutePath,
"--database",
desktopDb.absolutePath,
)
.contains("ACK_ACCEPT=acknowledged")
)
assertTrue(
desktopDestination.listSessions().any {
it.sessionId == "se_99999999-9999-4999-8999-999999999999"
}
)
} finally {
source.close();destination.close();desktopDestination.close();context.deleteDatabase(sourceName);context.deleteDatabase(destinationName);context.deleteDatabase(desktopDestinationName);root.deleteRecursively()
source.close()
destination.close()
desktopDestination.close()
context.deleteDatabase(sourceName)
context.deleteDatabase(destinationName)
context.deleteDatabase(desktopDestinationName)
root.deleteRecursively()
}
}
}

View file

@ -1,43 +1,137 @@
import json, os, signal, subprocess, tempfile, time, unittest, uuid
from pathlib import Path
ROOT=Path(__file__).resolve().parents[1]
ORCHESTRATOR=ROOT/"tools/sync_orchestrator.py"
def identity(prefix): return prefix+str(uuid.uuid4())
ROOT = Path(__file__).resolve().parents[1]
ORCHESTRATOR = ROOT / "tools/sync_orchestrator.py"
def identity(prefix):
return prefix + str(uuid.uuid4())
class OrchestratorTest(unittest.TestCase):
def fixture(self, root, **overrides):
db=root/"trainlog.db";db.touch();state=Path(str(db)+".sync-run.json")
run=identity("sy_");request=identity("sy_")
state.write_text(json.dumps({"run_id":run,"request_id":request,"trigger":"web","phase":"requested","progress_revision":1,"result":"running"}))
config={"format":"trainlog-sync-orchestrator-config","version":1,"enabled":True,"mode":"directory","expected_peer_id":identity("peer_"),"transport_root":str(root/"transport"),"owned_root":str(root/"owned"),"timeout_seconds":30}|overrides
(root/"transport").mkdir();(root/"owned").mkdir();path=root/"config.json";path.write_text(json.dumps(config))
return db,state,path,run,request
def invoke(self, executable, values):
db,state,config,run,request=values
return subprocess.run(["python3",str(executable),"--database",str(db),"--state",str(state),"--config",str(config),"--run-id",run,"--request-id",request],capture_output=True,text=True)
def test_browser_configuration_cannot_select_command_or_boolean_timeout(self):
for override in ({"peer_command":["evil"]},{"timeout_seconds":True},{"mode":"unsupported"}):
with self.subTest(override=override),tempfile.TemporaryDirectory() as directory:
root=Path(directory);values=self.fixture(root,**override);result=self.invoke(ORCHESTRATOR,values)
self.assertEqual(result.returncode,2);self.assertEqual(json.loads(values[1].read_text())["phase"],"failed")
def test_stdout_bound_is_enforced_before_complete_accumulation(self):
with tempfile.TemporaryDirectory() as directory:
root=Path(directory);sandbox=root/"program";sandbox.mkdir();(sandbox/"sync_orchestrator.py").write_bytes(ORCHESTRATOR.read_bytes())
(sandbox/"sync_peer_worker.py").write_text("import sys,time\nsys.stdout.write('x'*70000);sys.stdout.flush();time.sleep(30)\n")
values=self.fixture(root);result=self.invoke(sandbox/"sync_orchestrator.py",values)
self.assertEqual(result.returncode,2);final=json.loads(values[1].read_text());self.assertIn("exceeds 64 KiB",final["diagnostic"])
def test_shutdown_reaps_ignoring_helper_and_descendant(self):
with tempfile.TemporaryDirectory() as directory:
root=Path(directory);sandbox=root/"program";sandbox.mkdir();pidfile=root/"pids";(sandbox/"sync_orchestrator.py").write_bytes(ORCHESTRATOR.read_bytes())
(sandbox/"sync_peer_worker.py").write_text("import os,signal,time\nsignal.signal(signal.SIGTERM,signal.SIG_IGN)\npid=os.fork()\nif pid==0:\n signal.signal(signal.SIGTERM,signal.SIG_IGN);open(%r,'a').write(str(os.getpid())+'\\n');time.sleep(60)\nelse:\n open(%r,'a').write(str(os.getpid())+'\\n');time.sleep(60)\n"%(str(pidfile),str(pidfile)))
values=self.fixture(root);db,state,config,run,request=values
process=subprocess.Popen(["python3",str(sandbox/"sync_orchestrator.py"),"--database",str(db),"--state",str(state),"--config",str(config),"--run-id",run,"--request-id",request])
deadline=time.monotonic()+5
while not pidfile.is_file() and time.monotonic()<deadline: time.sleep(.02)
process.send_signal(signal.SIGTERM);self.assertEqual(process.wait(timeout=10),2)
for pid in map(int,pidfile.read_text().split()):
with self.assertRaises(ProcessLookupError): os.kill(pid,0)
self.assertEqual(json.loads(state.read_text())["phase"],"interrupted")
def fixture(self, root, **overrides):
db = root / "trainlog.db"
db.touch()
state = Path(str(db) + ".sync-run.json")
run = identity("sy_")
request = identity("sy_")
state.write_text(
json.dumps(
{
"run_id": run,
"request_id": request,
"trigger": "web",
"phase": "requested",
"progress_revision": 1,
"result": "running",
}
)
)
config = {
"format": "trainlog-sync-orchestrator-config",
"version": 1,
"enabled": True,
"mode": "directory",
"expected_peer_id": identity("peer_"),
"transport_root": str(root / "transport"),
"owned_root": str(root / "owned"),
"timeout_seconds": 30,
} | overrides
(root / "transport").mkdir()
(root / "owned").mkdir()
path = root / "config.json"
path.write_text(json.dumps(config))
return db, state, path, run, request
if __name__=="__main__": unittest.main()
def invoke(self, executable, values):
db, state, config, run, request = values
return subprocess.run(
[
"python3",
str(executable),
"--database",
str(db),
"--state",
str(state),
"--config",
str(config),
"--run-id",
run,
"--request-id",
request,
],
capture_output=True,
text=True,
)
def test_browser_configuration_cannot_select_command_or_boolean_timeout(self):
for override in (
{"peer_command": ["evil"]},
{"timeout_seconds": True},
{"mode": "unsupported"},
):
with self.subTest(override=override), tempfile.TemporaryDirectory() as directory:
root = Path(directory)
values = self.fixture(root, **override)
result = self.invoke(ORCHESTRATOR, values)
self.assertEqual(result.returncode, 2)
self.assertEqual(json.loads(values[1].read_text())["phase"], "failed")
def test_stdout_bound_is_enforced_before_complete_accumulation(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
sandbox = root / "program"
sandbox.mkdir()
(sandbox / "sync_orchestrator.py").write_bytes(ORCHESTRATOR.read_bytes())
(sandbox / "sync_peer_worker.py").write_text(
"import sys,time\nsys.stdout.write('x'*70000);sys.stdout.flush();time.sleep(30)\n"
)
values = self.fixture(root)
result = self.invoke(sandbox / "sync_orchestrator.py", values)
self.assertEqual(result.returncode, 2)
final = json.loads(values[1].read_text())
self.assertIn("exceeds 64 KiB", final["diagnostic"])
def test_shutdown_reaps_ignoring_helper_and_descendant(self):
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
sandbox = root / "program"
sandbox.mkdir()
pidfile = root / "pids"
(sandbox / "sync_orchestrator.py").write_bytes(ORCHESTRATOR.read_bytes())
(sandbox / "sync_peer_worker.py").write_text(
"import os,signal,time\nsignal.signal(signal.SIGTERM,signal.SIG_IGN)\npid=os.fork()\nif pid==0:\n signal.signal(signal.SIGTERM,signal.SIG_IGN);open(%r,'a').write(str(os.getpid())+'\\n');time.sleep(60)\nelse:\n open(%r,'a').write(str(os.getpid())+'\\n');time.sleep(60)\n"
% (str(pidfile), str(pidfile))
)
values = self.fixture(root)
db, state, config, run, request = values
process = subprocess.Popen(
[
"python3",
str(sandbox / "sync_orchestrator.py"),
"--database",
str(db),
"--state",
str(state),
"--config",
str(config),
"--run-id",
run,
"--request-id",
request,
]
)
deadline = time.monotonic() + 5
while not pidfile.is_file() and time.monotonic() < deadline:
time.sleep(0.02)
process.send_signal(signal.SIGTERM)
self.assertEqual(process.wait(timeout=10), 2)
for pid in map(int, pidfile.read_text().split()):
with self.assertRaises(ProcessLookupError):
os.kill(pid, 0)
self.assertEqual(json.loads(state.read_text())["phase"], "interrupted")
if __name__ == "__main__":
unittest.main()

View file

@ -1,83 +1,211 @@
#!/usr/bin/env python3
"""Real Firefox and Android production services in one correlated conversation."""
import json, os, signal, socket, sqlite3, subprocess, tempfile, time, unittest
from pathlib import Path
ROOT=Path(__file__).resolve().parents[1]
GECKO=Path(os.environ.get("TRAINLOG_GECKODRIVER","/home/fy59/.cache/trainlog/web-e2e-tools/geckodriver"))
ROOT = Path(__file__).resolve().parents[1]
GECKO = Path(
os.environ.get("TRAINLOG_GECKODRIVER", "/home/fy59/.cache/trainlog/web-e2e-tools/geckodriver")
)
def await_condition(predicate, seconds=30):
deadline=time.monotonic()+seconds
while time.monotonic()<deadline:
value=predicate()
if value: return value
time.sleep(.025)
raise AssertionError("condition did not become true")
deadline = time.monotonic() + seconds
while time.monotonic() < deadline:
value = predicate()
if value:
return value
time.sleep(0.025)
raise AssertionError("condition did not become true")
class BrowserSyncTest(unittest.TestCase):
def test_button_starts_worker_reconnects_and_displays_committed_draft(self):
if not GECKO.is_file(): self.skipTest("pinned geckodriver unavailable")
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.firefox.service import Service
from selenium.webdriver.support.ui import WebDriverWait
with tempfile.TemporaryDirectory(prefix="trainlog-web-sync-") as directory:
root=Path(directory);xdg=root/"data";xdg.mkdir();config=root/"config.json"
mode=os.environ.get("TRAINLOG_WEB_E2E_TRANSPORT_MODE","directory")
transport=root/"transport";transport.mkdir();owned=root/"owned";owned.mkdir()
remote=(root/"device/Documents/Trainlog") if mode=="mtp" else transport
remote.mkdir(parents=True,exist_ok=True)
release=root/"release-android";android_log=(root/"android.log").open("wb")
android_env=os.environ.copy();android_env.update({"JAVA_HOME":"/usr/lib/jvm/java-17-openjdk","TRAINLOG_SYNC_TRANSPORT_ROOT":str(remote),"TRAINLOG_SYNC_TEST_RELEASE_FILE":str(release)})
android=subprocess.Popen(["./gradlew","--no-daemon","testDebugUnitTest","--tests","com.labfytools.trainlog.data.SyncDeploymentConversationTest","--rerun-tasks"],cwd=ROOT/"android",env=android_env,stdout=android_log,stderr=subprocess.STDOUT,start_new_session=True)
try:
peer_file=await_condition(lambda: (remote/"android-peer-v1.json") if (remote/"android-peer-v1.json").is_file() else None,120)
except Exception as error:
if android.poll() is None: os.killpg(android.pid,signal.SIGTERM)
android.wait(timeout=10);android_log.close()
raise AssertionError(f"Android bridge did not advertise:\n{(root/'android.log').read_text(errors='replace')}") from error
peer=json.loads(peer_file.read_text())
config.write_text(json.dumps({"format":"trainlog-sync-orchestrator-config","version":1,"enabled":True,"mode":mode,"expected_peer_id":peer["peer_id"],"transport_root":str(transport),"owned_root":str(owned),"timeout_seconds":60}))
with socket.socket() as reserved:
reserved.bind(("127.0.0.1",0));port=reserved.getsockname()[1]
env=os.environ.copy();env.update({"XDG_DATA_HOME":str(xdg),"XDG_CONFIG_HOME":str(root/"config"),"XDG_CACHE_HOME":str(root/"cache"),"TRAINLOG_SYNC_GENERATION_CONFIG":str(config)})
if mode=="mtp": env.update({"TRAINLOG_SYNC_MTP_ADAPTER":str(ROOT/"build/tui/generation-mtp-adapter-double"),"TRAINLOG_MTP_DOUBLE_ROOT":str(root/"device")})
server=subprocess.Popen([str(ROOT/"build/tui/trainlog"),"--web","--port",str(port)],env=env,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL,start_new_session=True)
driver=None
try:
for _ in range(100):
with socket.socket() as probe:
if probe.connect_ex(("127.0.0.1",port))==0: break
time.sleep(.05)
options=webdriver.FirefoxOptions();options.add_argument("-headless")
driver=webdriver.Firefox(options=options,service=Service(str(GECKO)))
driver.get(f"http://127.0.0.1:{port}/")
wait=WebDriverWait(driver,15)
button=wait.until(lambda d:d.find_element(By.XPATH,"//button[normalize-space()='Synchroniser']"))
wait.until(lambda _ : button.is_enabled());button.click()
state_path=Path(str(xdg/"trainlog"/"trainlog.db")+".sync-run.json")
observed=await_condition(lambda: json.loads(state_path.read_text()) if state_path.is_file() and json.loads(state_path.read_text()).get("phase") in ("local_import_committed","failed","interrupted") else None,70)
self.assertEqual(observed["phase"],"local_import_committed",f"state={observed}\nandroid={(root/'android.log').read_text(errors='replace')}")
committed=observed
self.assertEqual(committed["result"],"running")
with sqlite3.connect(xdg/"trainlog"/"trainlog.db") as database:
self.assertEqual(database.execute("SELECT COUNT(*) FROM execution_drafts").fetchone()[0],1)
release.touch()
final=await_condition(lambda: json.loads(state_path.read_text()) if json.loads(state_path.read_text()).get("phase") in ("completed","failed","interrupted") else None)
self.assertEqual(final["phase"],"completed",f"state={final}\nandroid={((root/'android.log').read_text(errors='replace'))}")
wait.until(lambda d:"Synchronisation confirmée" in d.page_source)
summary=driver.find_element(By.XPATH,"//summary[contains(.,'Synchronisation confirmée')]");summary.click()
wait.until(lambda d:"Brouillons reçus" in d.page_source and "active" in d.page_source)
driver.refresh();wait.until(lambda d:"Synchronisation confirmée" in d.page_source)
state=json.loads(Path(str(xdg/"trainlog"/"trainlog.db")+".sync-run.json").read_text())
self.assertEqual(state["phase"],"completed")
self.assertEqual(android.wait(timeout=30),0,(root/"android.log").read_text(errors="replace"))
finally:
if driver is not None: driver.quit()
os.killpg(server.pid,signal.SIGTERM)
server.wait(timeout=10)
if android.poll() is None:
os.killpg(android.pid,signal.SIGTERM);android.wait(timeout=10)
android_log.close()
def test_button_starts_worker_reconnects_and_displays_committed_draft(self):
if not GECKO.is_file():
self.skipTest("pinned geckodriver unavailable")
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.firefox.service import Service
from selenium.webdriver.support.ui import WebDriverWait
if __name__=="__main__":unittest.main()
with tempfile.TemporaryDirectory(prefix="trainlog-web-sync-") as directory:
root = Path(directory)
xdg = root / "data"
xdg.mkdir()
config = root / "config.json"
mode = os.environ.get("TRAINLOG_WEB_E2E_TRANSPORT_MODE", "directory")
transport = root / "transport"
transport.mkdir()
owned = root / "owned"
owned.mkdir()
remote = (root / "device/Documents/Trainlog") if mode == "mtp" else transport
remote.mkdir(parents=True, exist_ok=True)
release = root / "release-android"
android_log = (root / "android.log").open("wb")
android_env = os.environ.copy()
android_env.update(
{
"JAVA_HOME": "/usr/lib/jvm/java-17-openjdk",
"TRAINLOG_SYNC_TRANSPORT_ROOT": str(remote),
"TRAINLOG_SYNC_TEST_RELEASE_FILE": str(release),
}
)
android = subprocess.Popen(
[
"./gradlew",
"--no-daemon",
"testDebugUnitTest",
"--tests",
"com.labfytools.trainlog.data.SyncDeploymentConversationTest",
"--rerun-tasks",
],
cwd=ROOT / "android",
env=android_env,
stdout=android_log,
stderr=subprocess.STDOUT,
start_new_session=True,
)
try:
peer_file = await_condition(
lambda: (
(remote / "android-peer-v1.json")
if (remote / "android-peer-v1.json").is_file()
else None
),
120,
)
except Exception as error:
if android.poll() is None:
os.killpg(android.pid, signal.SIGTERM)
android.wait(timeout=10)
android_log.close()
raise AssertionError(
f"Android bridge did not advertise:\n{(root/'android.log').read_text(errors='replace')}"
) from error
peer = json.loads(peer_file.read_text())
config.write_text(
json.dumps(
{
"format": "trainlog-sync-orchestrator-config",
"version": 1,
"enabled": True,
"mode": mode,
"expected_peer_id": peer["peer_id"],
"transport_root": str(transport),
"owned_root": str(owned),
"timeout_seconds": 60,
}
)
)
with socket.socket() as reserved:
reserved.bind(("127.0.0.1", 0))
port = reserved.getsockname()[1]
env = os.environ.copy()
env.update(
{
"XDG_DATA_HOME": str(xdg),
"XDG_CONFIG_HOME": str(root / "config"),
"XDG_CACHE_HOME": str(root / "cache"),
"TRAINLOG_SYNC_GENERATION_CONFIG": str(config),
}
)
if mode == "mtp":
env.update(
{
"TRAINLOG_SYNC_MTP_ADAPTER": str(
ROOT / "build/tui/generation-mtp-adapter-double"
),
"TRAINLOG_MTP_DOUBLE_ROOT": str(root / "device"),
}
)
server = subprocess.Popen(
[str(ROOT / "build/tui/trainlog"), "--web", "--port", str(port)],
env=env,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
driver = None
try:
for _ in range(100):
with socket.socket() as probe:
if probe.connect_ex(("127.0.0.1", port)) == 0:
break
time.sleep(0.05)
options = webdriver.FirefoxOptions()
options.add_argument("-headless")
driver = webdriver.Firefox(options=options, service=Service(str(GECKO)))
driver.get(f"http://127.0.0.1:{port}/")
wait = WebDriverWait(driver, 15)
button = wait.until(
lambda d: d.find_element(By.XPATH, "//button[normalize-space()='Synchroniser']")
)
wait.until(lambda _: button.is_enabled())
button.click()
state_path = Path(str(xdg / "trainlog" / "trainlog.db") + ".sync-run.json")
observed = await_condition(
lambda: (
json.loads(state_path.read_text())
if state_path.is_file()
and json.loads(state_path.read_text()).get("phase")
in ("local_import_committed", "failed", "interrupted")
else None
),
70,
)
self.assertEqual(
observed["phase"],
"local_import_committed",
f"state={observed}\nandroid={(root/'android.log').read_text(errors='replace')}",
)
committed = observed
self.assertEqual(committed["result"], "running")
with sqlite3.connect(xdg / "trainlog" / "trainlog.db") as database:
self.assertEqual(
database.execute("SELECT COUNT(*) FROM execution_drafts").fetchone()[0], 1
)
release.touch()
final = await_condition(
lambda: (
json.loads(state_path.read_text())
if json.loads(state_path.read_text()).get("phase")
in ("completed", "failed", "interrupted")
else None
)
)
self.assertEqual(
final["phase"],
"completed",
f"state={final}\nandroid={((root/'android.log').read_text(errors='replace'))}",
)
wait.until(lambda d: "Synchronisation confirmée" in d.page_source)
summary = driver.find_element(
By.XPATH, "//summary[contains(.,'Synchronisation confirmée')]"
)
summary.click()
wait.until(
lambda d: "Brouillons reçus" in d.page_source and "active" in d.page_source
)
driver.refresh()
wait.until(lambda d: "Synchronisation confirmée" in d.page_source)
state = json.loads(
Path(str(xdg / "trainlog" / "trainlog.db") + ".sync-run.json").read_text()
)
self.assertEqual(state["phase"], "completed")
self.assertEqual(
android.wait(timeout=30), 0, (root / "android.log").read_text(errors="replace")
)
finally:
if driver is not None:
driver.quit()
os.killpg(server.pid, signal.SIGTERM)
server.wait(timeout=10)
if android.poll() is None:
os.killpg(android.pid, signal.SIGTERM)
android.wait(timeout=10)
android_log.close()
if __name__ == "__main__":
unittest.main()

View file

@ -12,26 +12,70 @@ class BackupBridgeActivity : Activity() {
private lateinit var repository: TrainlogRepository
private lateinit var status: TextView
private val service by lazy { AndroidBackupService(applicationContext) }
override fun onCreate(state: Bundle?) {
super.onCreate(state); repository = TrainlogRepository(applicationContext)
status = TextView(this).apply { text = "Plaintext local backup. Keep the exported file private." }
val backup = Button(this).apply { text = "Create complete backup"; setOnClickListener {
startActivityForResult(Intent(Intent.ACTION_CREATE_DOCUMENT).setType("application/zip")
.putExtra(Intent.EXTRA_TITLE, "trainlog-backup.tlbackup").addCategory(Intent.CATEGORY_OPENABLE), 1) } }
val restore = Button(this).apply { text = "Verify and restore backup"; setOnClickListener {
startActivityForResult(Intent(Intent.ACTION_OPEN_DOCUMENT).setType("application/zip")
.addCategory(Intent.CATEGORY_OPENABLE), 2) } }
setContentView(LinearLayout(this).apply { orientation = LinearLayout.VERTICAL; addView(status); addView(backup); addView(restore) })
super.onCreate(state)
repository = TrainlogRepository(applicationContext)
status =
TextView(this).apply {
text = "Plaintext local backup. Keep the exported file private."
}
val backup =
Button(this).apply {
text = "Create complete backup"
setOnClickListener {
startActivityForResult(
Intent(Intent.ACTION_CREATE_DOCUMENT)
.setType("application/zip")
.putExtra(Intent.EXTRA_TITLE, "trainlog-backup.tlbackup")
.addCategory(Intent.CATEGORY_OPENABLE),
1,
)
}
}
val restore =
Button(this).apply {
text = "Verify and restore backup"
setOnClickListener {
startActivityForResult(
Intent(Intent.ACTION_OPEN_DOCUMENT)
.setType("application/zip")
.addCategory(Intent.CATEGORY_OPENABLE),
2,
)
}
}
setContentView(
LinearLayout(this).apply {
orientation = LinearLayout.VERTICAL
addView(status)
addView(backup)
addView(restore)
}
)
}
override fun onActivityResult(request: Int, result: Int, data: Intent?) {
super.onActivityResult(request, result, data); if (result != RESULT_OK || data?.data == null) return
if (request == 1) contentResolver.openOutputStream(data.data!!, "w")!!.use {
status.text = service.create(repository, it).toString()
} else if (request == 2) contentResolver.openInputStream(data.data!!)!!.use {
val verified = service.verify(it).getOrElse { error -> status.text = error.message; return }
status.text = service.restore(repository, verified).toString()
recreate()
}
super.onActivityResult(request, result, data)
if (result != RESULT_OK || data?.data == null) return
if (request == 1)
contentResolver.openOutputStream(data.data!!, "w")!!.use {
status.text = service.create(repository, it).toString()
}
else if (request == 2)
contentResolver.openInputStream(data.data!!)!!.use {
val verified =
service.verify(it).getOrElse { error ->
status.text = error.message
return
}
status.text = service.restore(repository, verified).toString()
recreate()
}
}
override fun onDestroy() {
repository.close()
super.onDestroy()
}
override fun onDestroy() { repository.close(); super.onDestroy() }
}

View file

@ -19,26 +19,75 @@ import org.junit.runner.RunWith
import org.robolectric.RobolectricTestRunner
import org.robolectric.annotation.Config
@RunWith(RobolectricTestRunner::class) @Config(sdk=[35])
@RunWith(RobolectricTestRunner::class)
@Config(sdk = [35])
class BackupBridgeServiceTest {
private val context: Context = ApplicationProvider.getApplicationContext()
@Test fun schemaSeventeenWalDraftRoundTripsWithoutMigration() {
val name="bridge-${UUID.randomUUID()}.db";var repository=TrainlogRepository(context,name);val service=AndroidBackupService(context)
@Test
fun schemaSeventeenWalDraftRoundTripsWithoutMigration() {
val name = "bridge-${UUID.randomUUID()}.db"
var repository = TrainlogRepository(context, name)
val service = AndroidBackupService(context)
try {
val exercise=(repository.createExercise(NewExerciseProfile("Bridge",RecordingMode.SETS,TrackingMode.REPS,0)) as CreateExerciseResult.Created).exercise
assertEquals(ActiveDraftMutationResult.Saved,repository.saveActiveSessionDraft(ActiveSessionDraft(
exercises=listOf(SessionExerciseDraft(exercise=exercise,sets=listOf(SessionSetDraft(reps=7,weightKg=20.0)))),
form=SessionDraftForm(selectedExercise=exercise,weightText="20,"))))
context.getDatabasePath(name).let { android.database.sqlite.SQLiteDatabase.openDatabase(it.path,null,android.database.sqlite.SQLiteDatabase.OPEN_READWRITE).use { db ->
db.enableWriteAheadLogging();db.execSQL("UPDATE active_session_draft SET weight_text='21,' WHERE id=1")
assertEquals(17,db.version)
} }
val bytes=ByteArrayOutputStream();assertTrue(service.create(repository,bytes) is AndroidBackupResult.Success)
System.getenv("TRAINLOG_BRIDGE_BACKUP_OUTPUT")?.let { java.io.File(it).writeBytes(bytes.toByteArray()) }
val verified=service.verify(ByteArrayInputStream(bytes.toByteArray())).getOrThrow()
assertTrue(service.restore(repository,verified,name) is AndroidBackupResult.Success)
repository=TrainlogRepository(context,name);assertEquals("21,",(repository.loadActiveSessionDraft() as ActiveDraftLoadResult.Loaded).draft.form.weightText)
context.getDatabasePath(name).let { android.database.sqlite.SQLiteDatabase.openDatabase(it.path,null,android.database.sqlite.SQLiteDatabase.OPEN_READONLY).use { db -> assertEquals(17,db.version) } }
} finally {repository.close();context.deleteDatabase(name)}
val exercise =
(repository.createExercise(
NewExerciseProfile("Bridge", RecordingMode.SETS, TrackingMode.REPS, 0)
) as CreateExerciseResult.Created)
.exercise
assertEquals(
ActiveDraftMutationResult.Saved,
repository.saveActiveSessionDraft(
ActiveSessionDraft(
exercises =
listOf(
SessionExerciseDraft(
exercise = exercise,
sets = listOf(SessionSetDraft(reps = 7, weightKg = 20.0)),
)
),
form = SessionDraftForm(selectedExercise = exercise, weightText = "20,"),
)
),
)
context.getDatabasePath(name).let {
android.database.sqlite.SQLiteDatabase.openDatabase(
it.path,
null,
android.database.sqlite.SQLiteDatabase.OPEN_READWRITE,
)
.use { db ->
db.enableWriteAheadLogging()
db.execSQL("UPDATE active_session_draft SET weight_text='21,' WHERE id=1")
assertEquals(17, db.version)
}
}
val bytes = ByteArrayOutputStream()
assertTrue(service.create(repository, bytes) is AndroidBackupResult.Success)
System.getenv("TRAINLOG_BRIDGE_BACKUP_OUTPUT")?.let {
java.io.File(it).writeBytes(bytes.toByteArray())
}
val verified = service.verify(ByteArrayInputStream(bytes.toByteArray())).getOrThrow()
assertTrue(service.restore(repository, verified, name) is AndroidBackupResult.Success)
repository = TrainlogRepository(context, name)
assertEquals(
"21,",
(repository.loadActiveSessionDraft() as ActiveDraftLoadResult.Loaded)
.draft
.form
.weightText,
)
context.getDatabasePath(name).let {
android.database.sqlite.SQLiteDatabase.openDatabase(
it.path,
null,
android.database.sqlite.SQLiteDatabase.OPEN_READONLY,
)
.use { db -> assertEquals(17, db.version) }
}
} finally {
repository.close()
context.deleteDatabase(name)
}
}
}

View file

@ -1,42 +1,112 @@
#!/usr/bin/env python3
"""Materialize and build the non-migrating schema-v17 Android backup bridge."""
import argparse, io, os, shutil, subprocess, tarfile
from pathlib import Path
ROOT=Path(__file__).resolve().parents[1]
BASE="77c15172e1401b91b8e2537c4b8548095ed50053"
ROOT = Path(__file__).resolve().parents[1]
BASE = "77c15172e1401b91b8e2537c4b8548095ed50053"
def replace(path: Path, old: str, new: str):
value = path.read_text()
if value.count(old) != 1:
raise RuntimeError(f"bridge patch anchor mismatch: {path}: {old[:40]}")
path.write_text(value.replace(old, new))
def replace(path:Path,old:str,new:str):
value=path.read_text()
if value.count(old)!=1: raise RuntimeError(f"bridge patch anchor mismatch: {path}: {old[:40]}")
path.write_text(value.replace(old,new))
def main():
parser=argparse.ArgumentParser();parser.add_argument("--output",required=True,type=Path);parser.add_argument("--jdk",default="/usr/lib/jvm/java-17-openjdk")
args=parser.parse_args()
if args.output.exists(): raise RuntimeError("bridge output already exists")
parser = argparse.ArgumentParser()
parser.add_argument("--output", required=True, type=Path)
parser.add_argument("--jdk", default="/usr/lib/jvm/java-17-openjdk")
args = parser.parse_args()
if args.output.exists():
raise RuntimeError("bridge output already exists")
args.output.mkdir(parents=True)
raw=subprocess.check_output(["git","archive","--format=tar",BASE,"android","catalog"],cwd=ROOT)
raw = subprocess.check_output(
["git", "archive", "--format=tar", BASE, "android", "catalog"], cwd=ROOT
)
with tarfile.open(fileobj=io.BytesIO(raw)) as archive:
for member in archive.getmembers():
target=(args.output/member.name).resolve()
if args.output.resolve() not in target.parents: raise RuntimeError("unsafe archive member")
archive.extractall(args.output,filter="data")
android=args.output/"android"
repository=android/"app/src/main/java/com/labfytools/trainlog/data/TrainlogRepository.kt"
replace(repository,"import java.text.Normalizer","import java.text.Normalizer\nimport java.io.File")
replace(repository," fun close() {\n database.close()\n }"," fun close() {\n database.close()\n }\n\n internal fun backupDatabaseTo(destination: File) {\n require(!destination.exists())\n val escaped = destination.absolutePath.replace(\"'\", \"''\")\n database.writableDatabase.execSQL(\"VACUUM INTO '$escaped'\")\n }")
shutil.copy2(ROOT/"android/app/src/main/java/com/labfytools/trainlog/data/AndroidBackupService.kt",android/"app/src/main/java/com/labfytools/trainlog/data/AndroidBackupService.kt")
shutil.copy2(ROOT/"tools/android_backup_bridge/BackupBridgeActivity.kt",android/"app/src/main/java/com/labfytools/trainlog/data/BackupBridgeActivity.kt")
shutil.copy2(ROOT/"tools/android_backup_bridge/BackupBridgeServiceTest.kt",android/"app/src/test/java/com/labfytools/trainlog/data/BackupBridgeServiceTest.kt")
manifest=android/"app/src/main/AndroidManifest.xml"
replace(manifest," <activity\n android:name=\".MainActivity\"", " <activity android:name=\".data.BackupBridgeActivity\" android:exported=\"true\" android:label=\"Trainlog Backup\">\n <intent-filter><action android:name=\"android.intent.action.MAIN\"/><category android:name=\"android.intent.category.LAUNCHER\"/></intent-filter>\n </activity>\n <activity\n android:name=\".MainActivity\"")
gradle=android/"app/build.gradle.kts"
replace(gradle,"versionCode = 3","versionCode = providers.environmentVariable(\"TRAINLOG_BRIDGE_VERSION_CODE\").orNull?.toInt() ?: 4")
env=os.environ.copy();env["JAVA_HOME"]=args.jdk;env["TRAINLOG_BRIDGE_VERSION_CODE"]=env.get("TRAINLOG_BRIDGE_VERSION_CODE","4");env["TRAINLOG_BRIDGE_BACKUP_OUTPUT"]=str(args.output/"schema17-synthetic.tlbackup")
subprocess.run([str(android/"gradlew"),"--no-daemon",":app:testDebugUnitTest","--tests","com.labfytools.trainlog.data.BackupBridgeServiceTest",":app:assembleDebug"],cwd=android,env=env,check=True)
migration_env=os.environ.copy();migration_env["JAVA_HOME"]=args.jdk;migration_env["TRAINLOG_BRIDGE_BACKUP_INPUT"]=env["TRAINLOG_BRIDGE_BACKUP_OUTPUT"]
subprocess.run([str(ROOT/"android/gradlew"),"--no-daemon",":app:testDebugUnitTest","--tests","com.labfytools.trainlog.data.BridgeBackupMigrationTest","--rerun-tasks"],cwd=ROOT/"android",env=migration_env,check=True)
print(f"BRIDGE_BASE={BASE}");print(f"BRIDGE_RECIPE={subprocess.check_output(['git','rev-parse','HEAD'],cwd=ROOT,text=True).strip()}")
target = (args.output / member.name).resolve()
if args.output.resolve() not in target.parents:
raise RuntimeError("unsafe archive member")
archive.extractall(args.output, filter="data")
android = args.output / "android"
repository = android / "app/src/main/java/com/labfytools/trainlog/data/TrainlogRepository.kt"
replace(
repository,
"import java.text.Normalizer",
"import java.text.Normalizer\nimport java.io.File",
)
replace(
repository,
" fun close() {\n database.close()\n }",
' fun close() {\n database.close()\n }\n\n internal fun backupDatabaseTo(destination: File) {\n require(!destination.exists())\n val escaped = destination.absolutePath.replace("\'", "\'\'")\n database.writableDatabase.execSQL("VACUUM INTO \'$escaped\'")\n }',
)
shutil.copy2(
ROOT / "android/app/src/main/java/com/labfytools/trainlog/data/AndroidBackupService.kt",
android / "app/src/main/java/com/labfytools/trainlog/data/AndroidBackupService.kt",
)
shutil.copy2(
ROOT / "tools/android_backup_bridge/BackupBridgeActivity.kt",
android / "app/src/main/java/com/labfytools/trainlog/data/BackupBridgeActivity.kt",
)
shutil.copy2(
ROOT / "tools/android_backup_bridge/BackupBridgeServiceTest.kt",
android / "app/src/test/java/com/labfytools/trainlog/data/BackupBridgeServiceTest.kt",
)
manifest = android / "app/src/main/AndroidManifest.xml"
replace(
manifest,
' <activity\n android:name=".MainActivity"',
' <activity android:name=".data.BackupBridgeActivity" android:exported="true" android:label="Trainlog Backup">\n <intent-filter><action android:name="android.intent.action.MAIN"/><category android:name="android.intent.category.LAUNCHER"/></intent-filter>\n </activity>\n <activity\n android:name=".MainActivity"',
)
gradle = android / "app/build.gradle.kts"
replace(
gradle,
"versionCode = 3",
'versionCode = providers.environmentVariable("TRAINLOG_BRIDGE_VERSION_CODE").orNull?.toInt() ?: 4',
)
env = os.environ.copy()
env["JAVA_HOME"] = args.jdk
env["TRAINLOG_BRIDGE_VERSION_CODE"] = env.get("TRAINLOG_BRIDGE_VERSION_CODE", "4")
env["TRAINLOG_BRIDGE_BACKUP_OUTPUT"] = str(args.output / "schema17-synthetic.tlbackup")
subprocess.run(
[
str(android / "gradlew"),
"--no-daemon",
":app:testDebugUnitTest",
"--tests",
"com.labfytools.trainlog.data.BackupBridgeServiceTest",
":app:assembleDebug",
],
cwd=android,
env=env,
check=True,
)
migration_env = os.environ.copy()
migration_env["JAVA_HOME"] = args.jdk
migration_env["TRAINLOG_BRIDGE_BACKUP_INPUT"] = env["TRAINLOG_BRIDGE_BACKUP_OUTPUT"]
subprocess.run(
[
str(ROOT / "android/gradlew"),
"--no-daemon",
":app:testDebugUnitTest",
"--tests",
"com.labfytools.trainlog.data.BridgeBackupMigrationTest",
"--rerun-tasks",
],
cwd=ROOT / "android",
env=migration_env,
check=True,
)
print(f"BRIDGE_BASE={BASE}")
print(
f"BRIDGE_RECIPE={subprocess.check_output(['git','rev-parse','HEAD'],cwd=ROOT,text=True).strip()}"
)
print(f"BRIDGE_APK={android/'app/build/outputs/apk/debug/app-debug.apk'}")
if __name__=="__main__": main()
if __name__ == "__main__":
main()

View file

@ -1,5 +1,6 @@
#!/usr/bin/env python3
"""Stage a reproducible, non-secret Trainlog desktop synchronization candidate."""
import argparse
import hashlib
import json
@ -10,15 +11,25 @@ from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
RUNTIME_TOOLS = [
"sync_orchestrator.py", "sync_peer_worker.py", "sync_generation_exchange.py",
"trainlog_sqlite.py", "validate_json.py", "import_mobile_export.py",
"import_exercise_aliases.py", "import_equipment_definitions.py",
"import_equipment_associations.py", "import_exercise_body_zones.py",
"import_training_feedback.py", "execution_draft_exchange.py",
"causal_delete_exchange.py", "import_exercise_profile_state.py",
"sync_orchestrator.py",
"sync_peer_worker.py",
"sync_generation_exchange.py",
"trainlog_sqlite.py",
"validate_json.py",
"import_mobile_export.py",
"import_exercise_aliases.py",
"import_equipment_definitions.py",
"import_equipment_associations.py",
"import_exercise_body_zones.py",
"import_training_feedback.py",
"execution_draft_exchange.py",
"causal_delete_exchange.py",
"import_exercise_profile_state.py",
"export_exercise_profile_state.py",
"export_pc_mobile.py", "export_equipment_definitions.py",
"export_exercise_body_zones.py", "export_training_feedback.py",
"export_pc_mobile.py",
"export_equipment_definitions.py",
"export_exercise_body_zones.py",
"export_training_feedback.py",
]
@ -54,20 +65,53 @@ def main() -> int:
for source in sorted((ROOT / "catalog").glob("*.json")):
shutil.copy2(source, args.output / "catalog" / source.name)
launcher = args.output / "bin/trainlog"
launcher.write_text("#!/bin/sh\nset -eu\nroot=$(CDPATH= cd -- \"$(dirname -- \"$0\")/..\" && pwd)\nexport TRAINLOG_SYNC_TOOLS_DIR=\"$root/tools\"\nexport TRAINLOG_SYNC_MTP_ADAPTER=\"$root/libexec/trainlog-generation-mtp-adapter\"\nexec \"$root/libexec/trainlog\" \"$@\"\n")
launcher.write_text(
'#!/bin/sh\nset -eu\nroot=$(CDPATH= cd -- "$(dirname -- "$0")/.." && pwd)\nexport TRAINLOG_SYNC_TOOLS_DIR="$root/tools"\nexport TRAINLOG_SYNC_MTP_ADAPTER="$root/libexec/trainlog-generation-mtp-adapter"\nexec "$root/libexec/trainlog" "$@"\n'
)
files = sorted(path for path in args.output.rglob("*") if path.is_file())
inventory = {
"format": "trainlog-sync-candidate-inventory", "version": 1,
"source_commit": subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip(),
"product_version": "0.1.2", "desktop_schema": 21, "android_schema": 20,
"protocols": ["mobile-export-v3", "mobile-history-v4", "causal-delete-v1",
"execution-draft-v1", "generation-manifest-v1", "generation-ack-v1"],
"entry_points": ["bin/trainlog", "libexec/trainlog-generation-mtp-adapter",
"tools/sync_orchestrator.py"],
"runtime_dependencies": ["python>=3.11", "sqlite3", "utf8proc", "libuuid", "libudev", "libmtp", "notcurses"],
"files": [{"path": str(path.relative_to(args.output)), "sha256": digest(path), "size": path.stat().st_size} for path in files],
"format": "trainlog-sync-candidate-inventory",
"version": 1,
"source_commit": subprocess.check_output(
["git", "rev-parse", "HEAD"], cwd=ROOT, text=True
).strip(),
"product_version": "0.1.2",
"desktop_schema": 21,
"android_schema": 20,
"protocols": [
"mobile-export-v3",
"mobile-history-v4",
"causal-delete-v1",
"execution-draft-v1",
"generation-manifest-v1",
"generation-ack-v1",
],
"entry_points": [
"bin/trainlog",
"libexec/trainlog-generation-mtp-adapter",
"tools/sync_orchestrator.py",
],
"runtime_dependencies": [
"python>=3.11",
"sqlite3",
"utf8proc",
"libuuid",
"libudev",
"libmtp",
"notcurses",
],
"files": [
{
"path": str(path.relative_to(args.output)),
"sha256": digest(path),
"size": path.stat().st_size,
}
for path in files
],
}
(args.output / "candidate-inventory.json").write_text(json.dumps(inventory, indent=2, sort_keys=True) + "\n")
(args.output / "candidate-inventory.json").write_text(
json.dumps(inventory, indent=2, sort_keys=True) + "\n"
)
os.chmod(args.output / "bin/trainlog", 0o755)
os.chmod(args.output / "libexec/trainlog", 0o755)
os.chmod(args.output / "libexec/trainlog-generation-mtp-adapter", 0o755)

View file

@ -5,6 +5,7 @@ The HTTP adapter starts this executable with fixed arguments. Peer transport is
selected only by a trusted local configuration file; request JSON can never
choose commands, paths, databases, capabilities, or causal policy.
"""
from __future__ import annotations
import argparse
@ -25,9 +26,19 @@ from pathlib import Path
MAX_CONFIG = 16 * 1024
MAX_REPORT = 64 * 1024
MAX_DIAGNOSTIC = 1024
PHASES = {"requested", "waiting_android_publication", "running",
"local_import_committed", "published", "waiting_acknowledgement",
"peer_consumed", "completed", "failed", "interrupted", "explicitly_degraded"}
PHASES = {
"requested",
"waiting_android_publication",
"running",
"local_import_committed",
"published",
"waiting_acknowledgement",
"peer_consumed",
"completed",
"failed",
"interrupted",
"explicitly_degraded",
}
class OrchestratorInterrupted(RuntimeError):
@ -40,8 +51,10 @@ _owned_process_group: int | None = None
def _forward_termination(signum: int, _frame: object) -> None:
"""Forward shutdown only to the exactly-owned helper process group."""
if _owned_process_group is not None:
try: os.killpg(_owned_process_group, signal.SIGTERM)
except ProcessLookupError: pass
try:
os.killpg(_owned_process_group, signal.SIGTERM)
except ProcessLookupError:
pass
raise OrchestratorInterrupted(f"orchestrator received signal {signum}")
@ -54,8 +67,9 @@ def _child_setup() -> None:
def canonical(value: object) -> bytes:
return (json.dumps(value, ensure_ascii=False, sort_keys=True,
separators=(",", ":")) + "\n").encode()
return (
json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n"
).encode()
def atomic_write(path: Path, value: dict) -> None:
@ -67,97 +81,176 @@ def atomic_write(path: Path, value: dict) -> None:
try:
os.fchmod(fd, 0o600)
with os.fdopen(fd, "wb") as stream:
stream.write(raw); stream.flush(); os.fsync(stream.fileno())
stream.write(raw)
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, path)
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
try: os.fsync(directory)
finally: os.close(directory)
try:
os.fsync(directory)
finally:
os.close(directory)
finally:
try: os.unlink(temporary)
except FileNotFoundError: pass
try:
os.unlink(temporary)
except FileNotFoundError:
pass
def load_json(path: Path, limit: int) -> dict:
with path.open("rb") as stream:
raw = stream.read(limit + 1)
if len(raw) > limit: raise RuntimeError(f"{path.name} exceeds its bound")
if len(raw) > limit:
raise RuntimeError(f"{path.name} exceeds its bound")
value = json.loads(raw.decode("utf-8", "strict"))
if not isinstance(value, dict): raise RuntimeError(f"{path.name} must contain an object")
if not isinstance(value, dict):
raise RuntimeError(f"{path.name} must contain an object")
return value
def update(state_path: Path, state: dict, phase: str, **fields: object) -> None:
if phase not in PHASES: raise RuntimeError("invalid run phase")
state.update(fields); state["phase"] = phase
if phase not in PHASES:
raise RuntimeError("invalid run phase")
state.update(fields)
state["phase"] = phase
state["progress_revision"] = int(state.get("progress_revision", 0)) + 1
state["updated_at"] = datetime.now(timezone.utc).isoformat()
atomic_write(state_path, state)
def validate_result(value: dict, run_id: str) -> dict:
required = {"format", "version", "run_id", "producer_peer_id", "consumer_peer_id",
"inbound_generation_id", "outbound_generation_id", "manifest_sha256",
"result", "sessions_reconciled", "domains", "drafts", "ai_midpoint", "ai_post_sync"}
if set(value) != required or value.get("format") != "trainlog-sync-worker-report" or value.get("version") != 1:
required = {
"format",
"version",
"run_id",
"producer_peer_id",
"consumer_peer_id",
"inbound_generation_id",
"outbound_generation_id",
"manifest_sha256",
"result",
"sessions_reconciled",
"domains",
"drafts",
"ai_midpoint",
"ai_post_sync",
}
if (
set(value) != required
or value.get("format") != "trainlog-sync-worker-report"
or value.get("version") != 1
):
raise RuntimeError("malformed peer worker report")
if value.get("run_id") != run_id or value.get("result") != "completed":
raise RuntimeError("peer worker did not complete the correlated run")
for identity, prefix in ((value["producer_peer_id"], "peer_"),
(value["consumer_peer_id"], "peer_"),
(value["inbound_generation_id"], "gen_"),
(value["outbound_generation_id"], "gen_")):
if not isinstance(identity, str) or not re.fullmatch(re.escape(prefix)+r"[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}",identity):
for identity, prefix in (
(value["producer_peer_id"], "peer_"),
(value["consumer_peer_id"], "peer_"),
(value["inbound_generation_id"], "gen_"),
(value["outbound_generation_id"], "gen_"),
):
if not isinstance(identity, str) or not re.fullmatch(
re.escape(prefix)
+ r"[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}",
identity,
):
raise RuntimeError("invalid worker identity")
if type(value["sessions_reconciled"]) is not int or value["sessions_reconciled"] < 0:
raise RuntimeError("invalid sessions_reconciled")
if not isinstance(value["domains"], dict) or not all(isinstance(k, str) and isinstance(v, bool) for k, v in value["domains"].items()):
if not isinstance(value["domains"], dict) or not all(
isinstance(k, str) and isinstance(v, bool) for k, v in value["domains"].items()
):
raise RuntimeError("invalid domain coverage")
if not isinstance(value["drafts"], list) or len(value["drafts"]) > 32:
raise RuntimeError("invalid draft summary")
return value
def read_worker(child: subprocess.Popen[bytes], state_path: Path, state: dict,
run_id: str, timeout: int) -> tuple[dict, bytes]:
def read_worker(
child: subprocess.Popen[bytes], state_path: Path, state: dict, run_id: str, timeout: int
) -> tuple[dict, bytes]:
selector = selectors.DefaultSelector()
assert child.stdout is not None and child.stderr is not None
selector.register(child.stdout, selectors.EVENT_READ, "stdout")
selector.register(child.stderr, selectors.EVENT_READ, "stderr")
buffers = {"stdout": bytearray(), "stderr": bytearray()}
line_buffer = bytearray(); final: dict | None = None
line_buffer = bytearray()
final: dict | None = None
deadline = time.monotonic() + timeout
allowed = ["waiting_android_publication","running","local_import_committed",
"published","waiting_acknowledgement","peer_consumed"]
allowed = [
"waiting_android_publication",
"running",
"local_import_committed",
"published",
"waiting_acknowledgement",
"peer_consumed",
]
last_index = -1
while selector.get_map():
remaining = deadline - time.monotonic()
if remaining <= 0: raise RuntimeError("peer worker timed out")
for key, _ in selector.select(min(remaining, .25)):
if remaining <= 0:
raise RuntimeError("peer worker timed out")
for key, _ in selector.select(min(remaining, 0.25)):
chunk = os.read(key.fileobj.fileno(), 4096)
if not chunk: selector.unregister(key.fileobj); continue
if not chunk:
selector.unregister(key.fileobj)
continue
target = buffers[key.data]
if len(target) + len(chunk) > MAX_REPORT: raise RuntimeError(f"peer worker {key.data} exceeds 64 KiB")
if len(target) + len(chunk) > MAX_REPORT:
raise RuntimeError(f"peer worker {key.data} exceeds 64 KiB")
target.extend(chunk)
if key.data == "stdout":
line_buffer.extend(chunk)
while b"\n" in line_buffer:
raw, _, rest = line_buffer.partition(b"\n"); line_buffer[:] = rest
if len(raw) > 8192: raise RuntimeError("progress frame exceeds 8 KiB")
raw, _, rest = line_buffer.partition(b"\n")
line_buffer[:] = rest
if len(raw) > 8192:
raise RuntimeError("progress frame exceeds 8 KiB")
value = json.loads(raw.decode("utf-8", "strict"))
if value.get("format") == "trainlog-sync-progress":
if set(value)-{"format","version","run_id","phase","producer_peer_id","consumer_peer_id","inbound_generation_id","outbound_generation_id","manifest_sha256"} or value.get("version") != 1 or value.get("run_id") != run_id or value.get("phase") not in allowed:
if (
set(value)
- {
"format",
"version",
"run_id",
"phase",
"producer_peer_id",
"consumer_peer_id",
"inbound_generation_id",
"outbound_generation_id",
"manifest_sha256",
}
or value.get("version") != 1
or value.get("run_id") != run_id
or value.get("phase") not in allowed
):
raise RuntimeError("invalid progress frame")
index=allowed.index(value["phase"])
if index < last_index: raise RuntimeError("progress phase regressed")
last_index=index;update(state_path,state,value["phase"],**{k:v for k,v in value.items() if k not in {"format","version","run_id","phase"}})
else: final=value
index = allowed.index(value["phase"])
if index < last_index:
raise RuntimeError("progress phase regressed")
last_index = index
update(
state_path,
state,
value["phase"],
**{
k: v
for k, v in value.items()
if k not in {"format", "version", "run_id", "phase"}
},
)
else:
final = value
child.wait()
if line_buffer.strip(): raise RuntimeError("unterminated worker frame")
if line_buffer.strip():
raise RuntimeError("unterminated worker frame")
if child.returncode != 0:
detail=bytes(buffers["stderr"]).decode("utf-8","replace").strip()[:MAX_DIAGNOSTIC]
detail = bytes(buffers["stderr"]).decode("utf-8", "replace").strip()[:MAX_DIAGNOSTIC]
raise RuntimeError(f"peer worker exited {child.returncode}: {detail}")
if final is None: raise RuntimeError("peer worker omitted final report")
return validate_result(final,run_id), bytes(buffers["stderr"])
if final is None:
raise RuntimeError("peer worker omitted final report")
return validate_result(final, run_id), bytes(buffers["stderr"])
def run(args: argparse.Namespace) -> int:
@ -167,15 +260,33 @@ def run(args: argparse.Namespace) -> int:
config = load_json(config_path, MAX_CONFIG)
if state.get("run_id") != args.run_id or state.get("request_id") != args.request_id:
raise RuntimeError("admission identity mismatch")
if set(config) != {"format", "version", "enabled", "mode", "expected_peer_id", "transport_root", "owned_root", "timeout_seconds"}:
if set(config) != {
"format",
"version",
"enabled",
"mode",
"expected_peer_id",
"transport_root",
"owned_root",
"timeout_seconds",
}:
raise RuntimeError("invalid trusted sync configuration")
if config["format"] != "trainlog-sync-orchestrator-config" or config["version"] != 1 or config["enabled"] is not True:
if (
config["format"] != "trainlog-sync-orchestrator-config"
or config["version"] != 1
or config["enabled"] is not True
):
raise RuntimeError("full-generation synchronization is disabled")
if config["mode"] not in ("directory", "mtp"):
raise RuntimeError("invalid transport mode")
for key in ("transport_root","owned_root"):
if not isinstance(config[key],str) or not Path(config[key]).is_absolute(): raise RuntimeError("trusted paths must be absolute")
if not isinstance(config["expected_peer_id"],str) or not re.fullmatch(r"peer_[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}",config["expected_peer_id"]): raise RuntimeError("invalid expected peer identity")
for key in ("transport_root", "owned_root"):
if not isinstance(config[key], str) or not Path(config[key]).is_absolute():
raise RuntimeError("trusted paths must be absolute")
if not isinstance(config["expected_peer_id"], str) or not re.fullmatch(
r"peer_[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}",
config["expected_peer_id"],
):
raise RuntimeError("invalid expected peer identity")
timeout = config["timeout_seconds"]
if not isinstance(timeout, int) or timeout < 1 or timeout > 900:
raise RuntimeError("invalid worker timeout")
@ -184,39 +295,96 @@ def run(args: argparse.Namespace) -> int:
lock_path = Path(args.database).parent / "sync.lock"
lock_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600)
try:
try: fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
try:
fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
update(state_path, state, "failed", result="conflict", diagnostic="Another synchronization owns the common lock", finished_at=datetime.now(timezone.utc).isoformat())
update(
state_path,
state,
"failed",
result="conflict",
diagnostic="Another synchronization owns the common lock",
finished_at=datetime.now(timezone.utc).isoformat(),
)
return 4
update(state_path, state, "running", result="running", diagnostic="")
environment = os.environ.copy()
environment.update({"TRAINLOG_SYNC_RUN_ID": args.run_id,
"TRAINLOG_SYNC_TRIGGER": state["trigger"],
"TRAINLOG_SYNC_DATABASE": args.database})
command=[sys.executable,str(Path(__file__).with_name("sync_peer_worker.py")),"--database",args.database,"--transport-root",config["transport_root"],"--owned-root",config["owned_root"],"--run-id",args.run_id,"--expected-peer",config["expected_peer_id"],"--timeout",str(timeout),"--mode",config["mode"]]
environment.update(
{
"TRAINLOG_SYNC_RUN_ID": args.run_id,
"TRAINLOG_SYNC_TRIGGER": state["trigger"],
"TRAINLOG_SYNC_DATABASE": args.database,
}
)
command = [
sys.executable,
str(Path(__file__).with_name("sync_peer_worker.py")),
"--database",
args.database,
"--transport-root",
config["transport_root"],
"--owned-root",
config["owned_root"],
"--run-id",
args.run_id,
"--expected-peer",
config["expected_peer_id"],
"--timeout",
str(timeout),
"--mode",
config["mode"],
]
if config["mode"] == "mtp":
adapter = os.environ.get("TRAINLOG_SYNC_MTP_ADAPTER",
str(Path(__file__).resolve().parents[1] / "build/tui/trainlog-generation-mtp-adapter"))
adapter = os.environ.get(
"TRAINLOG_SYNC_MTP_ADAPTER",
str(
Path(__file__).resolve().parents[1]
/ "build/tui/trainlog-generation-mtp-adapter"
),
)
if not Path(adapter).is_absolute() or not os.access(adapter, os.X_OK):
raise RuntimeError("fixed MTP adapter executable is unavailable")
command.extend(["--mtp-adapter", adapter])
child = subprocess.Popen(command, stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=environment,
preexec_fn=_child_setup)
child = subprocess.Popen(
command,
stdin=subprocess.DEVNULL,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=environment,
preexec_fn=_child_setup,
)
_owned_process_group = child.pid
try: result,_ = read_worker(child,state_path,state,args.run_id,timeout)
try:
result, _ = read_worker(child, state_path, state, args.run_id, timeout)
except Exception:
if child.poll() is None:
os.killpg(child.pid,signal.SIGTERM)
try: child.wait(5)
except subprocess.TimeoutExpired: os.killpg(child.pid,signal.SIGKILL);child.wait()
os.killpg(child.pid, signal.SIGTERM)
try:
child.wait(5)
except subprocess.TimeoutExpired:
os.killpg(child.pid, signal.SIGKILL)
child.wait()
raise
finally:
_owned_process_group = None
update(state_path, state, "completed", result="completed", diagnostic="",
inbound_generation_id=result["inbound_generation_id"],outbound_generation_id=result["outbound_generation_id"],producer_peer_id=result["producer_peer_id"],consumer_peer_id=result["consumer_peer_id"],sessions_reconciled=result["sessions_reconciled"],domains=result["domains"],drafts=result["drafts"],manifest_sha256=result["manifest_sha256"],
ai_midpoint=result["ai_midpoint"], ai_post_sync=result["ai_post_sync"],
finished_at=datetime.now(timezone.utc).isoformat())
update(
state_path,
state,
"completed",
result="completed",
diagnostic="",
inbound_generation_id=result["inbound_generation_id"],
outbound_generation_id=result["outbound_generation_id"],
producer_peer_id=result["producer_peer_id"],
consumer_peer_id=result["consumer_peer_id"],
sessions_reconciled=result["sessions_reconciled"],
domains=result["domains"],
drafts=result["drafts"],
manifest_sha256=result["manifest_sha256"],
ai_midpoint=result["ai_midpoint"],
ai_post_sync=result["ai_post_sync"],
finished_at=datetime.now(timezone.utc).isoformat(),
)
return 0
finally:
os.close(lock_fd)
@ -224,21 +392,39 @@ def run(args: argparse.Namespace) -> int:
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--database", required=True); parser.add_argument("--state", required=True)
parser.add_argument("--config", required=True); parser.add_argument("--run-id", required=True)
parser.add_argument("--database", required=True)
parser.add_argument("--state", required=True)
parser.add_argument("--config", required=True)
parser.add_argument("--run-id", required=True)
parser.add_argument("--request-id", required=True)
args = parser.parse_args()
signal.signal(signal.SIGTERM, _forward_termination)
signal.signal(signal.SIGINT, _forward_termination)
try: return run(args)
except (OSError, ValueError, json.JSONDecodeError, RuntimeError, subprocess.TimeoutExpired) as error:
try:
return run(args)
except (
OSError,
ValueError,
json.JSONDecodeError,
RuntimeError,
subprocess.TimeoutExpired,
) as error:
try:
path = Path(args.state); state = load_json(path, MAX_REPORT)
path = Path(args.state)
state = load_json(path, MAX_REPORT)
phase = "interrupted" if isinstance(error, OrchestratorInterrupted) else "failed"
update(path, state, phase, result=phase, diagnostic=str(error)[:MAX_DIAGNOSTIC],
finished_at=datetime.now(timezone.utc).isoformat())
except Exception: pass
update(
path,
state,
phase,
result=phase,
diagnostic=str(error)[:MAX_DIAGNOSTIC],
finished_at=datetime.now(timezone.utc).isoformat(),
)
except Exception:
pass
return 2
if __name__ == "__main__": raise SystemExit(main())
if __name__ == "__main__":
raise SystemExit(main())

View file

@ -5,6 +5,7 @@ The directory is the object-level seam implemented by the MTP adapter in a
deployed bundle and directly supplied by isolated tests. Business documents are
always produced and consumed by the existing generation services.
"""
from __future__ import annotations
import argparse, hashlib, json, os, sys, time, uuid
from contextlib import closing, redirect_stdout
@ -13,107 +14,266 @@ from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent))
import sync_generation_exchange as generation
CAPS={"generation-manifest-v1","generation-ack-v1","causal-delete-v1","mobile-history-v4","execution-draft-v1"}
CAPS = {
"generation-manifest-v1",
"generation-ack-v1",
"causal-delete-v1",
"mobile-history-v4",
"execution-draft-v1",
}
def emit(run_id:str,phase:str,**values:object)->None:
value={"format":"trainlog-sync-progress","version":1,"run_id":run_id,"phase":phase}|values
print(json.dumps(value,sort_keys=True,separators=(",",":")),flush=True)
def bounded(path:Path,limit:int=65536)->bytes:
def emit(run_id: str, phase: str, **values: object) -> None:
value = {
"format": "trainlog-sync-progress",
"version": 1,
"run_id": run_id,
"phase": phase,
} | values
print(json.dumps(value, sort_keys=True, separators=(",", ":")), flush=True)
def bounded(path: Path, limit: int = 65536) -> bytes:
with path.open("rb") as stream:
raw=stream.read(limit+1)
if len(raw)>limit: raise RuntimeError(f"{path.name} exceeds bound")
raw = stream.read(limit + 1)
if len(raw) > limit:
raise RuntimeError(f"{path.name} exceeds bound")
return raw
def publish_json(path:Path,value:dict)->None:
"""Publish a coordination object only after all bytes are durable."""
temporary=path.with_name("."+path.name+".tmp-"+str(os.getpid()))
with temporary.open("wb") as stream:
stream.write(generation.canonical(value));stream.flush();os.fsync(stream.fileno())
os.chmod(temporary,0o600);os.replace(temporary,path)
def run_adapter(adapter:Path,operation:str,peer:str,root:Path,deadline:float,allow_missing_peer:bool=False)->bool:
remaining=max(1,int(deadline-time.monotonic()))
result=__import__("subprocess").run([str(adapter),operation,peer,str(root)],stdin=__import__("subprocess").DEVNULL,
capture_output=True,timeout=min(remaining,30),check=False)
if result.returncode!=0:
diagnostic=(result.stderr.decode("utf-8","replace")[:1024] or "MTP adapter failed").strip()
if allow_missing_peer and "expected peer was not found" in diagnostic: return False
def publish_json(path: Path, value: dict) -> None:
"""Publish a coordination object only after all bytes are durable."""
temporary = path.with_name("." + path.name + ".tmp-" + str(os.getpid()))
with temporary.open("wb") as stream:
stream.write(generation.canonical(value))
stream.flush()
os.fsync(stream.fileno())
os.chmod(temporary, 0o600)
os.replace(temporary, path)
def run_adapter(
adapter: Path,
operation: str,
peer: str,
root: Path,
deadline: float,
allow_missing_peer: bool = False,
) -> bool:
remaining = max(1, int(deadline - time.monotonic()))
result = __import__("subprocess").run(
[str(adapter), operation, peer, str(root)],
stdin=__import__("subprocess").DEVNULL,
capture_output=True,
timeout=min(remaining, 30),
check=False,
)
if result.returncode != 0:
diagnostic = (
result.stderr.decode("utf-8", "replace")[:1024] or "MTP adapter failed"
).strip()
if allow_missing_peer and "expected peer was not found" in diagnostic:
return False
raise RuntimeError(diagnostic)
return True
def wait_file(path:Path,deadline:float,pump=None)->None:
def wait_file(path: Path, deadline: float, pump=None) -> None:
while not path.is_file():
if time.monotonic()>=deadline: raise RuntimeError(f"timeout waiting for {path.name}")
if pump is not None: pump()
time.sleep(.05)
if time.monotonic() >= deadline:
raise RuntimeError(f"timeout waiting for {path.name}")
if pump is not None:
pump()
time.sleep(0.05)
def wait_json(path:Path,deadline:float,predicate,pump=None)->dict:
def wait_json(path: Path, deadline: float, predicate, pump=None) -> dict:
while True:
wait_file(path,deadline,pump)
try: value=json.loads(bounded(path))
except (OSError,json.JSONDecodeError): value=None
if isinstance(value,dict) and predicate(value): return value
if time.monotonic()>=deadline: raise RuntimeError(f"timeout waiting for correlated {path.name}")
if pump is not None: pump()
time.sleep(.05)
wait_file(path, deadline, pump)
try:
value = json.loads(bounded(path))
except (OSError, json.JSONDecodeError):
value = None
if isinstance(value, dict) and predicate(value):
return value
if time.monotonic() >= deadline:
raise RuntimeError(f"timeout waiting for correlated {path.name}")
if pump is not None:
pump()
time.sleep(0.05)
def load_peer(root:Path,expected:str)->dict:
value=json.loads(bounded(root/"android-peer-v1.json"))
if set(value)!={"format","version","peer_id","capabilities"} or value["format"]!="trainlog-sync-peer" or type(value["version"]) is not int or value["version"]!=1 or value["peer_id"]!=expected or not isinstance(value["capabilities"],list) or not CAPS.issubset(set(value["capabilities"])):
def load_peer(root: Path, expected: str) -> dict:
value = json.loads(bounded(root / "android-peer-v1.json"))
if (
set(value) != {"format", "version", "peer_id", "capabilities"}
or value["format"] != "trainlog-sync-peer"
or type(value["version"]) is not int
or value["version"] != 1
or value["peer_id"] != expected
or not isinstance(value["capabilities"], list)
or not CAPS.issubset(set(value["capabilities"]))
):
raise RuntimeError("incompatible or unpaired Android peer advertisement")
return value
def main()->int:
parser=argparse.ArgumentParser();parser.add_argument("--database",required=True,type=Path);parser.add_argument("--transport-root",required=True,type=Path);parser.add_argument("--owned-root",required=True,type=Path);parser.add_argument("--run-id",required=True);parser.add_argument("--expected-peer",required=True);parser.add_argument("--timeout",required=True,type=int);parser.add_argument("--mode",choices=("directory","mtp"),default="directory");parser.add_argument("--mtp-adapter",type=Path)
args=parser.parse_args();deadline=time.monotonic()+args.timeout
pump=None
if args.mode=="mtp":
if args.mtp_adapter is None: raise RuntimeError("MTP adapter is required")
pump=lambda:run_adapter(args.mtp_adapter,"pull",args.expected_peer,args.transport_root,deadline,True)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--database", required=True, type=Path)
parser.add_argument("--transport-root", required=True, type=Path)
parser.add_argument("--owned-root", required=True, type=Path)
parser.add_argument("--run-id", required=True)
parser.add_argument("--expected-peer", required=True)
parser.add_argument("--timeout", required=True, type=int)
parser.add_argument("--mode", choices=("directory", "mtp"), default="directory")
parser.add_argument("--mtp-adapter", type=Path)
args = parser.parse_args()
deadline = time.monotonic() + args.timeout
pump = None
if args.mode == "mtp":
if args.mtp_adapter is None:
raise RuntimeError("MTP adapter is required")
pump = lambda: run_adapter(
args.mtp_adapter, "pull", args.expected_peer, args.transport_root, deadline, True
)
# Android publishes its durable identity only while participating. A
# missing expected advertisement is therefore a bounded wait state;
# device, transport, ambiguity and malformed-object failures remain
# immediate and distinct adapter errors.
wait_file(args.transport_root/"android-peer-v1.json",deadline,pump)
peer=load_peer(args.transport_root,args.expected_peer)
wait_file(args.transport_root / "android-peer-v1.json", deadline, pump)
peer = load_peer(args.transport_root, args.expected_peer)
with closing(generation.connect_database(args.database)) as db:
generation.require_schema(db);desktop_peer=generation.peer_identity(db,"desktop");db.commit()
request={"format":"trainlog-sync-generation-request","version":1,"run_id":args.run_id,"desktop_peer_id":desktop_peer,"android_peer_id":peer["peer_id"],"capabilities":sorted(CAPS)}
request_path=args.transport_root/"request-v1.json";publish_json(request_path,request)
if args.mode=="mtp": run_adapter(args.mtp_adapter,"push",args.expected_peer,args.transport_root,deadline)
emit(args.run_id,"waiting_android_publication",producer_peer_id=peer["peer_id"],consumer_peer_id=desktop_peer)
inbound_ref=args.transport_root/"android-generation-v1.json"
inbound=wait_json(inbound_ref,deadline,lambda value:value.get("run_id")==args.run_id,pump);inbound_dir=args.transport_root/inbound["relative_path"]
ack=generation.consume_desktop(args.database,inbound_dir)
ack_path=args.transport_root/"desktop-consumption-ack-v1.json";publish_json(ack_path,ack)
if args.mode=="mtp": run_adapter(args.mtp_adapter,"push",args.expected_peer,args.transport_root,deadline)
if ack["result"]!="consumed": raise RuntimeError("desktop rejected Android generation")
emit(args.run_id,"local_import_committed",inbound_generation_id=ack["generation_id"],manifest_sha256=ack["manifest_sha256"])
outgoing_id="gen_"+str(uuid.uuid4())
generation.require_schema(db)
desktop_peer = generation.peer_identity(db, "desktop")
db.commit()
request = {
"format": "trainlog-sync-generation-request",
"version": 1,
"run_id": args.run_id,
"desktop_peer_id": desktop_peer,
"android_peer_id": peer["peer_id"],
"capabilities": sorted(CAPS),
}
request_path = args.transport_root / "request-v1.json"
publish_json(request_path, request)
if args.mode == "mtp":
run_adapter(args.mtp_adapter, "push", args.expected_peer, args.transport_root, deadline)
emit(
args.run_id,
"waiting_android_publication",
producer_peer_id=peer["peer_id"],
consumer_peer_id=desktop_peer,
)
inbound_ref = args.transport_root / "android-generation-v1.json"
inbound = wait_json(
inbound_ref, deadline, lambda value: value.get("run_id") == args.run_id, pump
)
inbound_dir = args.transport_root / inbound["relative_path"]
ack = generation.consume_desktop(args.database, inbound_dir)
ack_path = args.transport_root / "desktop-consumption-ack-v1.json"
publish_json(ack_path, ack)
if args.mode == "mtp":
run_adapter(args.mtp_adapter, "push", args.expected_peer, args.transport_root, deadline)
if ack["result"] != "consumed":
raise RuntimeError("desktop rejected Android generation")
emit(
args.run_id,
"local_import_committed",
inbound_generation_id=ack["generation_id"],
manifest_sha256=ack["manifest_sha256"],
)
outgoing_id = "gen_" + str(uuid.uuid4())
# Export codecs retain human diagnostics on stdout when invoked as CLIs.
# The worker stdout contract is typed NDJSON only, so route those bounded
# diagnostics to the separately bounded stderr channel.
with redirect_stdout(sys.stderr):
stage,manifest,digest=generation.capture_desktop(args.database,args.owned_root,peer["peer_id"],args.run_id,outgoing_id)
stage, manifest, digest = generation.capture_desktop(
args.database, args.owned_root, peer["peer_id"], args.run_id, outgoing_id
)
del stage
published=generation.publish(args.database,outgoing_id,args.transport_root/"desktop-objects")
ref={"format":"trainlog-sync-generation-reference","version":1,"run_id":args.run_id,"generation_id":outgoing_id,"manifest_sha256":digest,"relative_path":str(published.relative_to(args.transport_root))}
publish_json(args.transport_root/"desktop-generation-v1.json",ref)
if args.mode=="mtp": run_adapter(args.mtp_adapter,"push",args.expected_peer,args.transport_root,deadline)
emit(args.run_id,"published",outbound_generation_id=outgoing_id,manifest_sha256=digest)
emit(args.run_id,"waiting_acknowledgement",outbound_generation_id=outgoing_id)
android_ack=args.transport_root/"android-consumption-ack-v1.json"
android_ack_value=wait_json(android_ack,deadline,lambda value:value.get("run_id")==args.run_id and value.get("generation_id")==outgoing_id,pump)
result=generation.accept_ack(args.database,android_ack)
if result not in ("acknowledged","unchanged"): raise RuntimeError(f"Android did not durably consume desktop generation: {result}: {android_ack_value.get('diagnostic','')}")
published = generation.publish(
args.database, outgoing_id, args.transport_root / "desktop-objects"
)
ref = {
"format": "trainlog-sync-generation-reference",
"version": 1,
"run_id": args.run_id,
"generation_id": outgoing_id,
"manifest_sha256": digest,
"relative_path": str(published.relative_to(args.transport_root)),
}
publish_json(args.transport_root / "desktop-generation-v1.json", ref)
if args.mode == "mtp":
run_adapter(args.mtp_adapter, "push", args.expected_peer, args.transport_root, deadline)
emit(args.run_id, "published", outbound_generation_id=outgoing_id, manifest_sha256=digest)
emit(args.run_id, "waiting_acknowledgement", outbound_generation_id=outgoing_id)
android_ack = args.transport_root / "android-consumption-ack-v1.json"
android_ack_value = wait_json(
android_ack,
deadline,
lambda value: value.get("run_id") == args.run_id
and value.get("generation_id") == outgoing_id,
pump,
)
result = generation.accept_ack(args.database, android_ack)
if result not in ("acknowledged", "unchanged"):
raise RuntimeError(
f"Android did not durably consume desktop generation: {result}: {android_ack_value.get('diagnostic','')}"
)
with closing(generation.connect_database(args.database)) as db:
inbound_row=db.execute("SELECT result FROM sync_consumed_generations WHERE generation_id=?",(ack["generation_id"],)).fetchone()
outbound_row=db.execute("SELECT status FROM sync_generations WHERE generation_id=?",(outgoing_id,)).fetchone()
drafts=[{"draft_id":row[0],"state":row[1],"session_type":json.loads(row[2]).get("session_type","training"),"occurrence_count":len(json.loads(row[2]).get("exercises",[]))} for row in db.execute("SELECT session_id,state,payload_json FROM execution_drafts ORDER BY session_id LIMIT 32")]
if inbound_row!=("consumed",) or outbound_row!=("acknowledged",): raise RuntimeError("generation evidence did not corroborate completion")
emit(args.run_id,"peer_consumed",outbound_generation_id=outgoing_id)
report={"format":"trainlog-sync-worker-report","version":1,"run_id":args.run_id,"producer_peer_id":peer["peer_id"],"consumer_peer_id":desktop_peer,"inbound_generation_id":ack["generation_id"],"outbound_generation_id":outgoing_id,"manifest_sha256":digest,"result":"completed","sessions_reconciled":0,"domains":{name:True for name in ("history-v4","execution-draft-v1","causal-delete-v1","equipment","aliases","profiles","body-zones","feedback")},"drafts":drafts,"ai_midpoint":{"result":"not_configured"},"ai_post_sync":{"result":"not_configured"}}
print(json.dumps(report,sort_keys=True,separators=(",",":")),flush=True);return 0
inbound_row = db.execute(
"SELECT result FROM sync_consumed_generations WHERE generation_id=?",
(ack["generation_id"],),
).fetchone()
outbound_row = db.execute(
"SELECT status FROM sync_generations WHERE generation_id=?", (outgoing_id,)
).fetchone()
drafts = [
{
"draft_id": row[0],
"state": row[1],
"session_type": json.loads(row[2]).get("session_type", "training"),
"occurrence_count": len(json.loads(row[2]).get("exercises", [])),
}
for row in db.execute(
"SELECT session_id,state,payload_json FROM execution_drafts ORDER BY session_id LIMIT 32"
)
]
if inbound_row != ("consumed",) or outbound_row != ("acknowledged",):
raise RuntimeError("generation evidence did not corroborate completion")
emit(args.run_id, "peer_consumed", outbound_generation_id=outgoing_id)
report = {
"format": "trainlog-sync-worker-report",
"version": 1,
"run_id": args.run_id,
"producer_peer_id": peer["peer_id"],
"consumer_peer_id": desktop_peer,
"inbound_generation_id": ack["generation_id"],
"outbound_generation_id": outgoing_id,
"manifest_sha256": digest,
"result": "completed",
"sessions_reconciled": 0,
"domains": {
name: True
for name in (
"history-v4",
"execution-draft-v1",
"causal-delete-v1",
"equipment",
"aliases",
"profiles",
"body-zones",
"feedback",
)
},
"drafts": drafts,
"ai_midpoint": {"result": "not_configured"},
"ai_post_sync": {"result": "not_configured"},
}
print(json.dumps(report, sort_keys=True, separators=(",", ":")), flush=True)
return 0
if __name__=="__main__": raise SystemExit(main())
if __name__ == "__main__":
raise SystemExit(main())