diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/DriveSyncTransport.kt b/android/app/src/main/java/com/labfytools/trainlog/data/DriveSyncTransport.kt
index 99eec1a..7e5af45 100644
--- a/android/app/src/main/java/com/labfytools/trainlog/data/DriveSyncTransport.kt
+++ b/android/app/src/main/java/com/labfytools/trainlog/data/DriveSyncTransport.kt
@@ -291,10 +291,15 @@ class DriveSyncWorker(context: Context, parameters: WorkerParameters) :
transport.pull(local)
stage = "coordinate"
val coordinator = SyncGenerationCoordinator(repository)
+ val terminalLedger = BackgroundSyncRunLedger(applicationContext)
coordinator.publishPeer(local)
stage = "peer-publication"
transport.pushPeer(local)
- if (!coordinator.hasActionableRequest(local)) {
+ val actionability = coordinator.classifyBackgroundRequest(local, terminalLedger.terminal())
+ if (
+ actionability == BackgroundRequestActionability.NOT_ACTIONABLE ||
+ actionability == BackgroundRequestActionability.RESUMABLE_BUT_STALE
+ ) {
if (coordinator.hasReplayableAcknowledgement(local)) {
stage = "acknowledgement-replay"
transport.pushAcknowledgement(local)
@@ -322,8 +327,31 @@ class DriveSyncWorker(context: Context, parameters: WorkerParameters) :
stage = "acknowledgement"
transport.pushAcknowledgement(local)
},
+ intent = SyncConversationIntent.RESUME_BACKGROUND,
)
- if (result is ForegroundGenerationResult.Completed) Result.success() else Result.retry()
+ when (result) {
+ is ForegroundGenerationResult.Completed -> {
+ terminalLedger.recordTerminal(
+ result.runId,
+ coordinator.remoteEvidence(local, result.runId),
+ )
+ Result.success()
+ }
+ ForegroundGenerationResult.Busy,
+ is ForegroundGenerationResult.Cancelled -> Result.success()
+ is ForegroundGenerationResult.Superseded -> {
+ result.runId?.let {
+ terminalLedger.recordTerminal(it, coordinator.remoteEvidence(local, it))
+ }
+ Result.success()
+ }
+ is ForegroundGenerationResult.Failed -> {
+ result.runId?.let {
+ terminalLedger.recordTerminal(it, coordinator.remoteEvidence(local, it))
+ }
+ Result.retry()
+ }
+ }
} catch (error: Exception) {
/* SECURITY: stage and exception class are fixed application
* metadata. Provider messages may contain private URIs and are
diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundActionability.kt b/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundActionability.kt
new file mode 100644
index 0000000..b11c76c
--- /dev/null
+++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundActionability.kt
@@ -0,0 +1,44 @@
+package com.labfytools.trainlog.data
+
+import android.content.Context
+
+internal enum class BackgroundRequestActionability {
+ NEW_REQUEST,
+ NEW_REMOTE_EVIDENCE,
+ ACTIVE_HANDOFF,
+ RESUMABLE_BUT_STALE,
+ NOT_ACTIONABLE,
+}
+
+internal data class BackgroundTerminalRun(
+ val runId: String,
+ val remoteEvidence: String,
+)
+
+/** One-entry durable suppression ledger for the current background request. */
+internal class BackgroundSyncRunLedger(context: Context) {
+ private val preferences =
+ context.applicationContext.getSharedPreferences(PREFERENCES, Context.MODE_PRIVATE)
+
+ fun terminal(): BackgroundTerminalRun? {
+ val runId = preferences.getString(RUN_ID, null) ?: return null
+ return BackgroundTerminalRun(runId, preferences.getString(REMOTE_EVIDENCE, "").orEmpty())
+ }
+
+ fun recordTerminal(runId: String, remoteEvidence: String) {
+ /* CONTRACT: only the request currently present on the transport needs
+ * suppression. A replacement request has a distinct run_id and is
+ * immediately actionable, so this ledger remains permanently bounded
+ * to one record and is not a second synchronization database. */
+ preferences.edit()
+ .putString(RUN_ID, runId)
+ .putString(REMOTE_EVIDENCE, remoteEvidence)
+ .apply()
+ }
+
+ companion object {
+ private const val PREFERENCES = "trainlog-background-sync-terminal"
+ private const val RUN_ID = "run_id"
+ private const val REMOTE_EVIDENCE = "remote_evidence"
+ }
+}
diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundService.kt b/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundService.kt
index ba8095a..6d0a73b 100644
--- a/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundService.kt
+++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundService.kt
@@ -76,6 +76,7 @@ class SyncBackgroundService : Service() {
requireNotNull(repository),
AndroidMtpPublicationVisibility(applicationContext),
)
+ val terminalLedger = BackgroundSyncRunLedger(applicationContext)
while (!stopped.get() && BackgroundSyncSettings(this).enabled) {
val directory = canonicalExchangeDirectory()
/* WHY: waiting inside run() owns the process-wide conversation
@@ -84,8 +85,36 @@ class SyncBackgroundService : Service() {
* the coordinator still validates all protocol evidence.
* INVARIANT: no request means no business work and no lock. */
coordinator.publishPeer(directory)
- if (coordinator.hasActionableRequest(directory)) {
- coordinator.run(directory, Duration.ofMinutes(5))
+ val actionability =
+ coordinator.classifyBackgroundRequest(directory, terminalLedger.terminal())
+ if (
+ actionability == BackgroundRequestActionability.NEW_REQUEST ||
+ actionability == BackgroundRequestActionability.NEW_REMOTE_EVIDENCE ||
+ actionability == BackgroundRequestActionability.ACTIVE_HANDOFF
+ ) {
+ when (
+ val result = coordinator.run(
+ directory,
+ Duration.ofMinutes(5),
+ intent = SyncConversationIntent.RESUME_BACKGROUND,
+ )
+ ) {
+ is ForegroundGenerationResult.Completed ->
+ terminalLedger.recordTerminal(
+ result.runId,
+ coordinator.remoteEvidence(directory, result.runId),
+ )
+ is ForegroundGenerationResult.Failed ->
+ result.runId?.let {
+ terminalLedger.recordTerminal(it, coordinator.remoteEvidence(directory, it))
+ }
+ is ForegroundGenerationResult.Superseded ->
+ result.runId?.let {
+ terminalLedger.recordTerminal(it, coordinator.remoteEvidence(directory, it))
+ }
+ ForegroundGenerationResult.Busy,
+ is ForegroundGenerationResult.Cancelled -> Unit
+ }
}
if (!stopped.get()) runCatching { Thread.sleep(5_000) }
}
diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/SyncConversationArbiter.kt b/android/app/src/main/java/com/labfytools/trainlog/data/SyncConversationArbiter.kt
new file mode 100644
index 0000000..4c63c2d
--- /dev/null
+++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncConversationArbiter.kt
@@ -0,0 +1,116 @@
+package com.labfytools.trainlog.data
+
+import java.time.Duration
+import java.util.concurrent.locks.ReentrantLock
+import kotlin.concurrent.withLock
+
+internal enum class SyncConversationOwner {
+ FOREGROUND_EXPLICIT,
+ BACKGROUND_AUTOMATIC,
+}
+
+internal class SyncConversationYieldedException : Exception("conversation superseded by explicit sync")
+
+/** Process-wide ownership authority for Android generation conversations. */
+internal class SyncConversationArbiter {
+ private val lock = ReentrantLock()
+ private val ownershipChanged = lock.newCondition()
+ private var active: Lease? = null
+ private var pendingExplicit = 0
+
+ /**
+ * Announces user intent before the foreground service is refreshed.
+ *
+ * WHY: starting the service first allowed it to acquire the old binary
+ * lock and silently absorb the click. CONTRACT: an announcement prevents
+ * new background ownership and cooperatively supersedes an existing
+ * background lease. INVARIANT: ownership is still exclusive until that
+ * lease observes yield and releases in its finally block.
+ */
+ fun announceExplicit(): ExplicitClaim = lock.withLock {
+ pendingExplicit += 1
+ active?.takeIf { it.owner == SyncConversationOwner.BACKGROUND_AUTOMATIC }
+ ?.requestYield()
+ ownershipChanged.signalAll()
+ ExplicitClaim(this)
+ }
+
+ fun tryAcquireBackground(): Lease? = lock.withLock {
+ if (active != null || pendingExplicit > 0) return null
+ Lease(this, SyncConversationOwner.BACKGROUND_AUTOMATIC).also { active = it }
+ }
+
+ private fun acquireExplicit(claim: ExplicitClaim, timeout: Duration): Lease? {
+ return lock.withLock {
+ check(claim.announced)
+ var remaining = timeout.toNanos()
+ while (active != null) {
+ if (remaining <= 0L) {
+ releaseAnnouncement(claim)
+ return@withLock null
+ }
+ remaining = ownershipChanged.awaitNanos(remaining)
+ }
+ releaseAnnouncement(claim)
+ Lease(this, SyncConversationOwner.FOREGROUND_EXPLICIT).also { active = it }
+ }
+ }
+
+ private fun releaseAnnouncement(claim: ExplicitClaim) {
+ if (!claim.announced) return
+ claim.announced = false
+ pendingExplicit -= 1
+ check(pendingExplicit >= 0)
+ ownershipChanged.signalAll()
+ }
+
+ private fun release(lease: Lease) = lock.withLock {
+ if (active === lease) {
+ active = null
+ ownershipChanged.signalAll()
+ }
+ }
+
+ internal class ExplicitClaim internal constructor(private val arbiter: SyncConversationArbiter) :
+ AutoCloseable {
+ internal var announced = true
+ private var acquired = false
+
+ fun acquire(timeout: Duration): Lease? {
+ check(!acquired)
+ val lease = arbiter.acquireExplicit(this, timeout)
+ acquired = lease != null
+ return lease
+ }
+
+ override fun close() = arbiter.lock.withLock {
+ if (!acquired) arbiter.releaseAnnouncement(this)
+ }
+ }
+
+ internal class Lease internal constructor(
+ private val arbiter: SyncConversationArbiter,
+ val owner: SyncConversationOwner,
+ ) : AutoCloseable {
+ @Volatile private var yieldRequested = false
+ @Volatile private var closed = false
+
+ internal fun requestYield() {
+ yieldRequested = true
+ }
+
+ fun throwIfYieldRequested() {
+ if (yieldRequested) throw SyncConversationYieldedException()
+ }
+
+ override fun close() {
+ if (closed) return
+ closed = true
+ arbiter.release(this)
+ }
+ }
+
+ companion object {
+ val process = SyncConversationArbiter()
+ }
+}
diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationForegroundCoordinator.kt b/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationForegroundCoordinator.kt
index 8a9b2f7..41d78e3 100644
--- a/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationForegroundCoordinator.kt
+++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationForegroundCoordinator.kt
@@ -13,8 +13,15 @@ import org.json.JSONObject
internal sealed interface ForegroundGenerationResult {
data class Completed(val runId: String) : ForegroundGenerationResult
+ data object Busy : ForegroundGenerationResult
+ data class Superseded(val runId: String?) : ForegroundGenerationResult
+ data class Cancelled(val runId: String?) : ForegroundGenerationResult
+ data class Failed(val message: String, val runId: String?) : ForegroundGenerationResult
+}
- data class Error(val message: String) : ForegroundGenerationResult
+internal enum class SyncConversationIntent {
+ NEW_EXPLICIT,
+ RESUME_BACKGROUND,
}
internal fun interface SyncGenerationTrace {
@@ -43,6 +50,7 @@ internal class SyncGenerationCoordinator(
private val repository: TrainlogRepository,
private val visibility: MtpPublicationVisibility = ImmediateMtpPublicationVisibility,
private val trace: SyncGenerationTrace = AndroidSyncGenerationTrace,
+ private val arbiter: SyncConversationArbiter = SyncConversationArbiter.process,
) {
private val service = SyncGenerationService(repository)
@@ -54,14 +62,6 @@ internal class SyncGenerationCoordinator(
result: String? = null,
) = trace.record(runId, phase, artifact, generationId, result)
- companion object {
- /* WHY: UI and background triggers share one repository protocol and
- * must never create concurrent runs. CONTRACT: this process-wide lock
- * guards only one bounded conversation; SQLite remains the durable
- * replay authority after process restart. */
- private val conversationLock = java.util.concurrent.locks.ReentrantLock()
- }
-
internal fun resumableGeneration(runId: String, consumerPeerId: String? = null): CapturedSyncGeneration? =
repository.inSyncGenerationTransaction { db ->
db.rawQuery(
@@ -90,10 +90,10 @@ internal class SyncGenerationCoordinator(
}
}
- internal fun hasActionableRequest(directory: File): Boolean {
+ private fun request(directory: File): JSONObject? {
val path = File(directory, "request-v1.json")
- if (!path.isFile || path.length() !in 1..64 * 1024) return false
- val request = runCatching { JSONObject(path.readText()) }.getOrNull() ?: return false
+ if (!path.isFile || path.length() !in 1..64 * 1024) return null
+ val request = runCatching { JSONObject(path.readText()) }.getOrNull() ?: return null
val runId = request.optString("run_id")
val desktopPeerId = request.optString("desktop_peer_id")
if (
@@ -102,16 +102,64 @@ internal class SyncGenerationCoordinator(
!runId.matches(Regex("^sy_[0-9a-f-]{36}$")) ||
!desktopPeerId.matches(Regex("^peer_[0-9a-f-]{36}$")) ||
request.optString("android_peer_id") != service.peerId()
- ) return false
- if (resumableGeneration(runId, desktopPeerId) != null) return true
- return !repository.inSyncGenerationTransaction { db ->
+ ) return null
+ return request
+ }
+
+ private fun hasGeneration(runId: String): Boolean =
+ repository.inSyncGenerationTransaction { db ->
db.rawQuery(
"SELECT 1 FROM sync_generations WHERE run_id=? LIMIT 1",
arrayOf(runId),
).use { it.moveToFirst() }
}
+
+ internal fun remoteEvidence(directory: File, runId: String): String {
+ val evidence =
+ listOf(
+ Triple("desktop-consumption-ack-v1.json", "trainlog-sync-ack", 1),
+ Triple("desktop-generation-v1.json", "trainlog-sync-generation-reference", 1),
+ )
+ .mapNotNull { (name, format, version) ->
+ val path = File(directory, name)
+ if (!path.isFile || path.length() !in 1..64 * 1024) return@mapNotNull null
+ val raw = runCatching { path.readBytes() }.getOrNull() ?: return@mapNotNull null
+ val value = runCatching { JSONObject(raw.toString(Charsets.UTF_8)) }.getOrNull()
+ ?: return@mapNotNull null
+ if (
+ value.optString("format") != format ||
+ value.optInt("version", -1) != version ||
+ value.optString("run_id") != runId
+ ) return@mapNotNull null
+ "$name:${java.security.MessageDigest.getInstance("SHA-256").digest(raw)
+ .joinToString("") { "%02x".format(it) }}"
+ }
+ return evidence.sorted().joinToString("|")
}
+ internal fun classifyBackgroundRequest(
+ directory: File,
+ terminal: BackgroundTerminalRun?,
+ ): BackgroundRequestActionability {
+ val request = request(directory) ?: return BackgroundRequestActionability.NOT_ACTIONABLE
+ val runId = request.getString("run_id")
+ val desktopPeerId = request.getString("desktop_peer_id")
+ val resumable = resumableGeneration(runId, desktopPeerId) != null
+ if (terminal?.runId == runId) {
+ val currentEvidence = remoteEvidence(directory, runId)
+ return if (currentEvidence.isNotEmpty() && currentEvidence != terminal.remoteEvidence)
+ BackgroundRequestActionability.NEW_REMOTE_EVIDENCE
+ else if (resumable) BackgroundRequestActionability.RESUMABLE_BUT_STALE
+ else BackgroundRequestActionability.NOT_ACTIONABLE
+ }
+ if (!hasGeneration(runId)) return BackgroundRequestActionability.NEW_REQUEST
+ if (resumable) return BackgroundRequestActionability.ACTIVE_HANDOFF
+ return BackgroundRequestActionability.NOT_ACTIONABLE
+ }
+
+ internal fun currentRequestRunId(directory: File): String? =
+ request(directory)?.optString("run_id")
+
internal fun hasReplayableAcknowledgement(directory: File): Boolean {
val request =
runCatching { JSONObject(File(directory, "request-v1.json").readText()) }.getOrNull()
@@ -287,8 +335,10 @@ internal class SyncGenerationCoordinator(
deadline: Long,
predicate: (JSONObject) -> Boolean,
pump: (() -> Unit)? = null,
+ lease: SyncConversationArbiter.Lease,
): ByteArray {
while (true) {
+ lease.throwIfYieldRequested()
if (path.isFile && path.length() <= 64 * 1024) {
try {
val raw = path.readBytes()
@@ -312,11 +362,27 @@ internal class SyncGenerationCoordinator(
afterPublication: (() -> Unit)? = null,
pollTransport: (() -> Unit)? = null,
afterAcknowledgement: (() -> Unit)? = null,
+ intent: SyncConversationIntent = SyncConversationIntent.RESUME_BACKGROUND,
+ explicitClaim: SyncConversationArbiter.ExplicitClaim? = null,
): ForegroundGenerationResult {
- if (!conversationLock.tryLock()) {
- return ForegroundGenerationResult.Error("Synchronization already in progress.")
+ val claim =
+ if (intent == SyncConversationIntent.NEW_EXPLICIT)
+ explicitClaim ?: arbiter.announceExplicit()
+ else null
+ val lease = try {
+ claim?.acquire(timeout) ?: if (claim == null) arbiter.tryAcquireBackground() else null
+ } catch (_: InterruptedException) {
+ Thread.currentThread().interrupt()
+ claim?.close()
+ return ForegroundGenerationResult.Cancelled(null)
}
+ if (lease == null) {
+ claim?.close()
+ return ForegroundGenerationResult.Busy
+ }
+ var currentRunId: String? = null
return try {
+ val baselineRunId = request(directory)?.optString("run_id")
val peer = publishPeer(directory)
phase(null, "peer_published", "android-peer-v1.json")
/* WHY: trainlog-syncd is intentionally driven by the established
@@ -325,23 +391,23 @@ internal class SyncGenerationCoordinator(
* identity before its caller emits one fresh request signal.
* INVARIANT: a failed attempt's request, run, generations, and
* ACK evidence remain immutable; retry creates new correlation. */
- afterPeerPublication?.invoke()
- phase(null, "legacy_trigger_published", "trainlog-sync-request-v1.json")
+ afterPeerPublication?.let {
+ it.invoke()
+ phase(null, "legacy_trigger_published", "trainlog-sync-request-v1.json")
+ }
val deadline = System.nanoTime() + timeout.toNanos()
val requestRaw =
awaitCorrelated(File(directory, "request-v1.json"), deadline, { candidate ->
val runId = candidate.optString("run_id")
- runId.startsWith("sy_") &&
- (resumableGeneration(runId) != null ||
- !repository.inSyncGenerationTransaction { db ->
- db.rawQuery(
- "SELECT 1 FROM sync_generations WHERE run_id=? LIMIT 1",
- arrayOf(runId),
- ).use { it.moveToFirst() }
- })
- }, pollTransport)
+ runId.startsWith("sy_") && when (intent) {
+ SyncConversationIntent.NEW_EXPLICIT -> runId != baselineRunId
+ SyncConversationIntent.RESUME_BACKGROUND ->
+ resumableGeneration(runId) != null || !hasGeneration(runId)
+ }
+ }, pollTransport, lease)
val request = JSONObject(requestRaw.toString(Charsets.UTF_8))
val runId = request.getString("run_id")
+ currentRunId = runId
phase(runId, "generation_request_observed", "request-v1.json")
if (request.getString("android_peer_id") != peer)
throw SyncGenerationException("request targets another Android peer")
@@ -354,6 +420,7 @@ internal class SyncGenerationCoordinator(
it.optString("android_peer_id") == peer
},
pollTransport,
+ lease,
)
reconcileArchiveAcknowledgements(
archiveAckRaw,
@@ -370,6 +437,7 @@ internal class SyncGenerationCoordinator(
* an acknowledged outbound generation remains resumable only to
* finish its missing inbound half. */
phase(runId, "generation_capture_started")
+ lease.throwIfYieldRequested()
val captured =
resumableGeneration(
request.getString("run_id"),
@@ -379,11 +447,13 @@ internal class SyncGenerationCoordinator(
request.getString("desktop_peer_id"),
request.getString("run_id"),
)
+ lease.throwIfYieldRequested()
phase(runId, "generation_captured", generationId = captured.generationId)
val published = service.publish(captured, File(directory, "android-objects"))
visibility.confirm(
published.walkTopDown().filter(File::isFile).sortedBy { it.name == "manifest.json" }.toList()
)
+ lease.throwIfYieldRequested()
phase(
runId,
"generation_objects_published",
@@ -410,6 +480,7 @@ internal class SyncGenerationCoordinator(
captured.generationId,
)
afterPublication?.invoke()
+ lease.throwIfYieldRequested()
// CONTRACT: durable objects from earlier conversations remain in
// the exchange directory. Only this run's exact generation ACK
// can advance its lineage; stale bytes are retained and ignored.
@@ -417,13 +488,13 @@ internal class SyncGenerationCoordinator(
awaitCorrelated(File(directory, "desktop-consumption-ack-v1.json"), deadline, {
it.optString("run_id") == captured.runId &&
it.optString("generation_id") == captured.generationId
- }, pollTransport)
+ }, pollTransport, lease)
service.acceptAcknowledgement(desktopAck)
phase(runId, "desktop_ack_observed", "desktop-consumption-ack-v1.json", captured.generationId)
val referenceRaw =
awaitCorrelated(File(directory, "desktop-generation-v1.json"), deadline, {
it.optString("run_id") == captured.runId
- }, pollTransport)
+ }, pollTransport, lease)
val reference = JSONObject(referenceRaw.toString(Charsets.UTF_8))
phase(
runId,
@@ -458,6 +529,13 @@ internal class SyncGenerationCoordinator(
afterAcknowledgement?.invoke()
phase(runId, "completed", result = "completed")
ForegroundGenerationResult.Completed(captured.runId)
+ } catch (_: SyncConversationYieldedException) {
+ phase(currentRunId, "superseded", result = "foreground_priority")
+ ForegroundGenerationResult.Superseded(currentRunId)
+ } catch (_: InterruptedException) {
+ Thread.currentThread().interrupt()
+ phase(currentRunId, "cancelled", result = "interrupted")
+ ForegroundGenerationResult.Cancelled(currentRunId)
} catch (error: Exception) {
if (error.message == "generation retention capacity exhausted") {
try {
@@ -486,9 +564,13 @@ internal class SyncGenerationCoordinator(
"error",
result = error.javaClass.simpleName,
)
- ForegroundGenerationResult.Error(error.message ?: "generation synchronization failed")
+ ForegroundGenerationResult.Failed(
+ error.message ?: "generation synchronization failed",
+ currentRunId,
+ )
} finally {
- conversationLock.unlock()
+ lease.close()
+ claim?.close()
}
}
}
diff --git a/android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt b/android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt
index 2e7e17a..11c7c4b 100644
--- a/android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt
+++ b/android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt
@@ -29,6 +29,9 @@ import com.labfytools.trainlog.data.SyncRequestOutbox
import com.labfytools.trainlog.data.SyncRequestResult
import com.labfytools.trainlog.data.TrainlogRepository
import com.labfytools.trainlog.data.ForegroundGenerationResult
+import com.labfytools.trainlog.data.BackgroundSyncRunLedger
+import com.labfytools.trainlog.data.SyncConversationArbiter
+import com.labfytools.trainlog.data.SyncConversationIntent
import com.labfytools.trainlog.data.SyncGenerationForegroundCoordinator
import com.labfytools.trainlog.data.BackgroundSyncSettings
import com.labfytools.trainlog.data.SyncBackgroundService
@@ -52,39 +55,72 @@ internal suspend fun publishBundleAndRequest(
exporter: SyncExporter,
requestOutbox: SyncRequestOutbox,
visibility: MtpPublicationVisibility = com.labfytools.trainlog.data.ImmediateMtpPublicationVisibility,
+ explicitClaim: SyncConversationArbiter.ExplicitClaim =
+ SyncConversationArbiter.process.announceExplicit(),
+ terminalLedger: BackgroundSyncRunLedger? = null,
): SyncRequestResult = withContext(Dispatchers.IO) {
- val opened = exporter.openStorageSnapshot()
- if (opened is com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Error) {
- return@withContext SyncRequestResult.Error("prepare:${opened.message}")
- }
- val snapshot = (opened as com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Ready).snapshot
- /* CONTRACT: the request signal remains compatible with an unconfigured
- * desktop, so publish the strict standalone V3 snapshot before admission.
- * A configured daemon still routes the signal exclusively to the complete
- * generation below; this compatibility publication never selects V3. */
- when (val publication = exporter.exportMobileBundle(snapshot)) {
- SyncExportResult.Unsupported -> return@withContext SyncRequestResult.Unsupported
- is SyncExportResult.Error ->
- return@withContext SyncRequestResult.Error("prepare:${publication.message}")
- is SyncExportResult.Exported -> Unit
- }
- return@withContext when (
- val result = SyncGenerationForegroundCoordinator(repository, visibility).run(
- canonicalExchangeDirectory(),
- afterPeerPublication = {
- when (val request = requestOutbox.requestSync(snapshot)) {
- is SyncRequestResult.Requested -> Unit
- SyncRequestResult.Unsupported ->
- throw IllegalStateException("Generation synchronization is unsupported.")
- is SyncRequestResult.Error -> throw IllegalStateException(request.message)
- is SyncRequestResult.Completed ->
- throw IllegalStateException("Unexpected completed request signal.")
+ try {
+ val opened = exporter.openStorageSnapshot()
+ if (opened is com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Error) {
+ return@withContext SyncRequestResult.Error("prepare:${opened.message}")
+ }
+ val snapshot =
+ (opened as com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Ready).snapshot
+ /* CONTRACT: the request signal remains compatible with an unconfigured
+ * desktop, so publish the strict standalone V3 snapshot before admission.
+ * A configured daemon still routes the signal exclusively to the complete
+ * generation below; this compatibility publication never selects V3. */
+ when (val publication = exporter.exportMobileBundle(snapshot)) {
+ SyncExportResult.Unsupported -> return@withContext SyncRequestResult.Unsupported
+ is SyncExportResult.Error ->
+ return@withContext SyncRequestResult.Error("prepare:${publication.message}")
+ is SyncExportResult.Exported -> Unit
+ }
+ val directory = canonicalExchangeDirectory()
+ val coordinator = SyncGenerationForegroundCoordinator(repository, visibility)
+ return@withContext when (
+ val result = coordinator.run(
+ directory,
+ afterPeerPublication = {
+ when (val request = requestOutbox.requestSync(snapshot)) {
+ is SyncRequestResult.Requested -> Unit
+ SyncRequestResult.Unsupported ->
+ throw IllegalStateException("Generation synchronization is unsupported.")
+ is SyncRequestResult.Error -> throw IllegalStateException(request.message)
+ is SyncRequestResult.Completed ->
+ throw IllegalStateException("Unexpected completed request signal.")
+ }
+ },
+ intent = SyncConversationIntent.NEW_EXPLICIT,
+ explicitClaim = explicitClaim,
+ )
+ ) {
+ is ForegroundGenerationResult.Completed -> SyncRequestResult.Completed(result.runId)
+ ForegroundGenerationResult.Busy ->
+ SyncRequestResult.Error("sync_interrupted")
+ is ForegroundGenerationResult.Superseded ->
+ SyncRequestResult.Error("sync_interrupted").also {
+ result.runId?.let { runId ->
+ terminalLedger?.recordTerminal(
+ runId,
+ coordinator.remoteEvidence(directory, runId),
+ )
+ }
}
- },
- )
- ) {
- is ForegroundGenerationResult.Completed -> SyncRequestResult.Completed(result.runId)
- is ForegroundGenerationResult.Error -> SyncRequestResult.Error(result.message)
+ is ForegroundGenerationResult.Cancelled ->
+ SyncRequestResult.Error("sync_interrupted")
+ is ForegroundGenerationResult.Failed ->
+ SyncRequestResult.Error(result.message).also {
+ result.runId?.let { runId ->
+ terminalLedger?.recordTerminal(
+ runId,
+ coordinator.remoteEvidence(directory, runId),
+ )
+ }
+ }
+ }
+ } finally {
+ explicitClaim.close()
}
}
@@ -147,8 +183,11 @@ fun SyncScreen(
add(strings.resources.getQuantityString(R.plurals.catalog_reconciled, reconciled, reconciled))
skipped?.let { add(strings.resources.getQuantityString(R.plurals.catalog_present, it, it)) }
}.joinToString(", ")
- fun visibleError(raw: String): String = raw.removePrefix("prepare:").let {
- if (raw.startsWith("prepare:")) strings.getString(R.string.sync_prepare_error, it) else raw
+ fun visibleError(raw: String): String = when {
+ raw == "sync_interrupted" -> strings.getString(R.string.sync_interrupted)
+ raw.startsWith("prepare:") ->
+ strings.getString(R.string.sync_prepare_error, raw.removePrefix("prepare:"))
+ else -> raw
}
var status by
@@ -393,15 +432,15 @@ fun SyncScreen(
* published from one prepared repository snapshot before
* that signal becomes visible; otherwise the PC can mix a
* fresh V3 file with stale or absent causal companions. */
- if (backgroundSettings.enabled) {
- /* WHY: Samsung may freeze the activity after USB mode
- * changes or a permission surface opens. CONTRACT: a
- * user-enabled foreground listener is refreshed before
- * the UI-owned conversation begins; the process lock
- * prevents it from creating a second generation. */
- SyncBackgroundService.start(context)
- }
+ /* WHY: Samsung may freeze the activity after USB mode
+ * changes or a permission surface opens. CONTRACT:
+ * announce explicit priority before waking the service.
+ * The service may keep Android alive, but cannot become
+ * conversation owner while this claim is pending. */
+ val explicitClaim = SyncConversationArbiter.process.announceExplicit()
+ if (backgroundSettings.enabled) SyncBackgroundService.start(context)
syncRunning = true
+ status = strings.getString(R.string.sync_running)
coroutineScope.launch {
try {
when (
@@ -410,6 +449,8 @@ fun SyncScreen(
exporter,
requestOutbox,
AndroidMtpPublicationVisibility(context),
+ explicitClaim,
+ BackgroundSyncRunLedger(context),
)
) {
is SyncRequestResult.Completed -> {
@@ -435,6 +476,7 @@ fun SyncScreen(
}
}
} finally {
+ explicitClaim.close()
syncRunning = false
}
}
diff --git a/android/app/src/main/res/values-en/strings.xml b/android/app/src/main/res/values-en/strings.xml
index 3f9891b..683de90 100644
--- a/android/app/src/main/res/values-en/strings.xml
+++ b/android/app/src/main/res/values-en/strings.xml
@@ -94,6 +94,7 @@
Synchronize
The Android snapshot is maintained automatically.
Synchronization in progress…
+ Synchronization interrupted. You can start a new attempt.
Synchronize now
Background synchronization
Allows Trainlog to receive PC synchronizations without opening the app.
diff --git a/android/app/src/main/res/values/strings.xml b/android/app/src/main/res/values/strings.xml
index 3a21413..bfc9fce 100644
--- a/android/app/src/main/res/values/strings.xml
+++ b/android/app/src/main/res/values/strings.xml
@@ -94,6 +94,7 @@
Synchroniser
Le snapshot Android est maintenu automatiquement.
Synchronisation en cours…
+ Synchronisation interrompue. Une nouvelle tentative peut être lancée.
Synchroniser maintenant
Synchronisation en arrière-plan
Permet à Trainlog de recevoir les synchronisations du PC sans ouvrir l’application.
diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SyncConversationOwnershipTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SyncConversationOwnershipTest.kt
new file mode 100644
index 0000000..e1babfd
--- /dev/null
+++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncConversationOwnershipTest.kt
@@ -0,0 +1,251 @@
+package com.labfytools.trainlog.data
+
+import android.content.Context
+import androidx.test.core.app.ApplicationProvider
+import java.io.File
+import java.nio.file.Files
+import java.time.Duration
+import java.util.UUID
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicReference
+import org.json.JSONArray
+import org.json.JSONObject
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertNotEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.robolectric.RobolectricTestRunner
+import org.robolectric.annotation.Config
+
+@RunWith(RobolectricTestRunner::class)
+@Config(sdk = [35])
+class SyncConversationOwnershipTest {
+ private val context: Context = ApplicationProvider.getApplicationContext()
+
+ private fun request(root: File, runId: String, androidPeer: String, desktopPeer: String) {
+ File(root, "request-v1.json").writeText(
+ JSONObject()
+ .put("format", "trainlog-sync-generation-request")
+ .put("version", 1)
+ .put("run_id", runId)
+ .put("android_peer_id", androidPeer)
+ .put("desktop_peer_id", desktopPeer)
+ .toString()
+ )
+ }
+
+ private fun archiveAcknowledgements(
+ root: File,
+ runId: String,
+ androidPeer: String,
+ desktopPeer: String,
+ ) {
+ File(root, "desktop-archive-acknowledgements-v1.json").writeText(
+ JSONObject()
+ .put("format", "trainlog-sync-archive-acknowledgements")
+ .put("version", 1)
+ .put("run_id", runId)
+ .put("android_peer_id", androidPeer)
+ .put("desktop_peer_id", desktopPeer)
+ .put("acknowledgements", JSONArray())
+ .toString()
+ )
+ }
+
+ @Test
+ fun explicitClickYieldsOldBackgroundAndPublishesFreshTriggerForNewRun() {
+ val databaseName = "generation-ownership-${UUID.randomUUID()}.db"
+ val root = Files.createTempDirectory("trainlog-generation-ownership-").toFile()
+ val repository = TrainlogRepository(context, databaseName)
+ val arbiter = SyncConversationArbiter()
+ val oldObserved = CountDownLatch(1)
+ val capturedRuns = mutableListOf()
+ val desktopPeer = "peer_${UUID.randomUUID()}"
+ val oldRun = "sy_${UUID.randomUUID()}"
+ val newRun = "sy_${UUID.randomUUID()}"
+ val service = SyncGenerationService(repository)
+ val background =
+ SyncGenerationCoordinator(
+ repository,
+ arbiter = arbiter,
+ trace = SyncGenerationTrace { runId, phase, _, _, _ ->
+ if (phase == "generation_request_observed" && runId == oldRun) oldObserved.countDown()
+ },
+ )
+ val foreground =
+ SyncGenerationCoordinator(
+ repository,
+ arbiter = arbiter,
+ trace = SyncGenerationTrace { runId, phase, _, _, _ ->
+ if (phase == "generation_captured") synchronized(capturedRuns) {
+ capturedRuns += checkNotNull(runId)
+ }
+ },
+ )
+ val executor = java.util.concurrent.Executors.newFixedThreadPool(2)
+ try {
+ request(root, oldRun, service.peerId(), desktopPeer)
+ File(root, "trainlog-sync-request-v1.json").writeText(
+ """{"format":"trainlog-sync-request","version":1,"request_id":"sr_old"}"""
+ )
+ val old = executor.submit {
+ background.run(root, Duration.ofSeconds(3))
+ }
+ assertTrue(oldObserved.await(2, TimeUnit.SECONDS))
+
+ val claim = arbiter.announceExplicit()
+ val snapshot =
+ (DirectExchangePublisher { directExchangeDirectoryForPath(root) }.snapshot()
+ as DirectExchangeSnapshotResult.Ready).snapshot
+ val outbox = SyncRequestOutbox { directExchangeDirectoryForPath(root) }
+ var freshRequestId = ""
+ val explicit = executor.submit {
+ foreground.run(
+ root,
+ Duration.ofMillis(700),
+ afterPeerPublication = {
+ freshRequestId =
+ (outbox.requestSync(snapshot) as SyncRequestResult.Requested).requestId
+ request(root, newRun, service.peerId(), desktopPeer)
+ archiveAcknowledgements(root, newRun, service.peerId(), desktopPeer)
+ },
+ intent = SyncConversationIntent.NEW_EXPLICIT,
+ explicitClaim = claim,
+ )
+ }
+
+ assertTrue(old.get(2, TimeUnit.SECONDS) is ForegroundGenerationResult.Superseded)
+ assertTrue(explicit.get(3, TimeUnit.SECONDS) is ForegroundGenerationResult.Failed)
+ assertNotEquals("sr_old", freshRequestId)
+ assertEquals(
+ freshRequestId,
+ JSONObject(File(root, "trainlog-sync-request-v1.json").readText())
+ .getString("request_id"),
+ )
+ assertEquals(listOf(newRun), capturedRuns)
+ repository.inSyncGenerationTransaction { database ->
+ database.rawQuery("SELECT run_id FROM sync_generations", null).use { cursor ->
+ assertTrue(cursor.moveToFirst())
+ assertEquals(newRun, cursor.getString(0))
+ assertEquals(false, cursor.moveToNext())
+ }
+ }
+ } finally {
+ executor.shutdownNow()
+ repository.close()
+ context.deleteDatabase(databaseName)
+ root.deleteRecursively()
+ }
+ }
+
+ @Test
+ fun terminalBackgroundRunStaysSuppressedUntilNewRequestOrRemoteEvidence() {
+ val databaseName = "generation-suppression-${UUID.randomUUID()}.db"
+ val root = Files.createTempDirectory("trainlog-generation-suppression-").toFile()
+ val repository = TrainlogRepository(context, databaseName)
+ try {
+ val service = SyncGenerationService(repository)
+ val coordinator = SyncGenerationCoordinator(repository, arbiter = SyncConversationArbiter())
+ val desktopPeer = "peer_${UUID.randomUUID()}"
+ val runId = "sy_${UUID.randomUUID()}"
+ request(root, runId, service.peerId(), desktopPeer)
+ service.capture(root, desktopPeer, runId)
+ val terminal = BackgroundTerminalRun(runId, coordinator.remoteEvidence(root, runId))
+
+ repeat(5) {
+ assertEquals(
+ BackgroundRequestActionability.RESUMABLE_BUT_STALE,
+ coordinator.classifyBackgroundRequest(root, terminal),
+ )
+ }
+
+ File(root, "desktop-consumption-ack-v1.json").writeText(
+ JSONObject()
+ .put("format", "trainlog-sync-ack")
+ .put("version", 1)
+ .put("run_id", runId)
+ .put("proof", "new")
+ .toString()
+ )
+ assertEquals(
+ BackgroundRequestActionability.NEW_REMOTE_EVIDENCE,
+ coordinator.classifyBackgroundRequest(root, terminal),
+ )
+
+ request(root, "sy_${UUID.randomUUID()}", service.peerId(), desktopPeer)
+ assertEquals(
+ BackgroundRequestActionability.NEW_REQUEST,
+ coordinator.classifyBackgroundRequest(root, terminal),
+ )
+ } finally {
+ repository.close()
+ context.deleteDatabase(databaseName)
+ root.deleteRecursively()
+ }
+ }
+
+ @Test
+ fun cancelledExplicitConversationHandsSameGenerationToBackgroundWithoutDuplication() {
+ val databaseName = "generation-handoff-${UUID.randomUUID()}.db"
+ val root = Files.createTempDirectory("trainlog-generation-handoff-").toFile()
+ val repository = TrainlogRepository(context, databaseName)
+ val arbiter = SyncConversationArbiter()
+ val published = CountDownLatch(1)
+ val result = AtomicReference()
+ val desktopPeer = "peer_${UUID.randomUUID()}"
+ val oldRunId = "sy_${UUID.randomUUID()}"
+ val runId = "sy_${UUID.randomUUID()}"
+ try {
+ val service = SyncGenerationService(repository)
+ request(root, oldRunId, service.peerId(), desktopPeer)
+ val coordinator =
+ SyncGenerationCoordinator(
+ repository,
+ arbiter = arbiter,
+ trace = SyncGenerationTrace { _, phase, _, _, _ ->
+ if (phase == "generation_reference_published") published.countDown()
+ },
+ )
+ val claim = arbiter.announceExplicit()
+ val thread = Thread {
+ result.set(
+ coordinator.run(
+ root,
+ Duration.ofSeconds(5),
+ afterPeerPublication = {
+ request(root, runId, service.peerId(), desktopPeer)
+ archiveAcknowledgements(root, runId, service.peerId(), desktopPeer)
+ },
+ intent = SyncConversationIntent.NEW_EXPLICIT,
+ explicitClaim = claim,
+ )
+ )
+ }
+ thread.start()
+ assertTrue(published.await(3, TimeUnit.SECONDS))
+ thread.interrupt()
+ thread.join(3_000)
+ assertTrue(result.get() is ForegroundGenerationResult.Cancelled)
+ assertEquals(
+ BackgroundRequestActionability.ACTIVE_HANDOFF,
+ coordinator.classifyBackgroundRequest(root, null),
+ )
+
+ val resumed = coordinator.run(root, Duration.ofMillis(250))
+ assertTrue(resumed is ForegroundGenerationResult.Failed)
+ repository.inSyncGenerationTransaction { database ->
+ database.rawQuery("SELECT COUNT(*) FROM sync_generations WHERE run_id=?", arrayOf(runId))
+ .use { cursor ->
+ assertTrue(cursor.moveToFirst())
+ assertEquals(1, cursor.getInt(0))
+ }
+ }
+ } finally {
+ repository.close()
+ context.deleteDatabase(databaseName)
+ root.deleteRecursively()
+ }
+ }
+}
diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationConversationLockTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationConversationLockTest.kt
index ba71b83..de31766 100644
--- a/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationConversationLockTest.kt
+++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationConversationLockTest.kt
@@ -38,12 +38,8 @@ class SyncGenerationConversationLockTest {
}
assertTrue(entered.await(2, TimeUnit.SECONDS))
val duplicate = background.run(root, Duration.ofMillis(10))
- assertTrue(duplicate is ForegroundGenerationResult.Error)
- assertEquals(
- "Synchronization already in progress.",
- (duplicate as ForegroundGenerationResult.Error).message,
- )
- assertTrue(active.get(2, TimeUnit.SECONDS) is ForegroundGenerationResult.Error)
+ assertEquals(ForegroundGenerationResult.Busy, duplicate)
+ assertTrue(active.get(2, TimeUnit.SECONDS) is ForegroundGenerationResult.Failed)
repository.inSyncGenerationTransaction { database ->
database.rawQuery("SELECT COUNT(*) FROM sync_generations", null).use { cursor ->
assertTrue(cursor.moveToFirst())
diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt
index 785d5ee..7cf221c 100644
--- a/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt
+++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt
@@ -357,9 +357,15 @@ class SyncGenerationServiceTest {
val repository = TrainlogRepository(context, databaseName)
try {
val coordinator = SyncGenerationCoordinator(repository)
- assertEquals(false, coordinator.hasActionableRequest(root))
+ assertEquals(
+ BackgroundRequestActionability.NOT_ACTIONABLE,
+ coordinator.classifyBackgroundRequest(root, null),
+ )
File(root, "request-v1.json").writeText("not-json")
- assertEquals(false, coordinator.hasActionableRequest(root))
+ assertEquals(
+ BackgroundRequestActionability.NOT_ACTIONABLE,
+ coordinator.classifyBackgroundRequest(root, null),
+ )
val peer = SyncGenerationService(repository).peerId()
val request =
@@ -370,11 +376,17 @@ class SyncGenerationServiceTest {
.put("android_peer_id", "peer_${UUID.randomUUID()}")
.put("desktop_peer_id", "peer_${UUID.randomUUID()}")
File(root, "request-v1.json").writeText(request.toString())
- assertEquals(false, coordinator.hasActionableRequest(root))
+ assertEquals(
+ BackgroundRequestActionability.NOT_ACTIONABLE,
+ coordinator.classifyBackgroundRequest(root, null),
+ )
request.put("android_peer_id", peer)
File(root, "request-v1.json").writeText(request.toString())
- assertEquals(true, coordinator.hasActionableRequest(root))
+ assertEquals(
+ BackgroundRequestActionability.NEW_REQUEST,
+ coordinator.classifyBackgroundRequest(root, null),
+ )
} finally {
repository.close()
context.deleteDatabase(databaseName)
diff --git a/tests/test_syncd_routing.py b/tests/test_syncd_routing.py
index f2b80d0..c05ec7f 100644
--- a/tests/test_syncd_routing.py
+++ b/tests/test_syncd_routing.py
@@ -74,6 +74,58 @@ class SyncdRoutingTest(unittest.TestCase):
self.assertEqual(trigger, state["trigger"])
self.assertEqual("full_generation_v1", state["effective_mode"])
+ def test_consumed_request_id_does_not_create_a_second_desktop_run(self):
+ with tempfile.TemporaryDirectory() as directory:
+ root = Path(directory)
+ transport = root / "transport"
+ transport.mkdir()
+ request = "sr_22222222-2222-4222-8222-222222222222"
+ config = root / "config.json"
+ config.write_text(
+ json.dumps(
+ {
+ "format": "trainlog-sync-orchestrator-config",
+ "version": 1,
+ "enabled": True,
+ "mode": "mtp",
+ "expected_peer_id": "peer_11111111-1111-4111-8111-111111111111",
+ "transport_root": str(transport),
+ "owned_root": str(root / "owned"),
+ "timeout_seconds": 30,
+ }
+ )
+ )
+ state = root / "trainlog.db.sync-run.json"
+ state.write_text(json.dumps({"request_id": request, "run_id": "sy_old"}))
+ args = argparse.Namespace(
+ mtp_adapter=root / "adapter",
+ state=state,
+ trigger="android",
+ orchestrator=root / "sync_orchestrator.py",
+ database=root / "trainlog.db",
+ )
+ commands = []
+
+ def execute(command, **_kwargs):
+ commands.append(command)
+ (transport / syncd.REQUEST_NAME).write_text(
+ json.dumps(
+ {
+ "format": "trainlog-sync-request",
+ "version": 1,
+ "request_id": request,
+ "requested_at": "2026-09-20T12:00:00Z",
+ }
+ )
+ )
+ return subprocess.CompletedProcess(command, 0, "PASS\n", "")
+
+ with mock.patch.object(syncd.subprocess, "run", side_effect=execute):
+ self.assertEqual(3, syncd.run_full_generation(args, config))
+ self.assertEqual(str(args.mtp_adapter), commands[0][0])
+ self.assertEqual(1, len(commands))
+ self.assertEqual("sy_old", json.loads(state.read_text())["run_id"])
+
def test_android_production_trigger_no_longer_uses_opt_in_or_legacy_export(self):
source = (ROOT / "android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt").read_text()
production = source.split("internal suspend fun publishBundleAndRequest", 1)[1].split(