fix(sync): recover lost generation acknowledgements

android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationForegroundCoordinator.kt

@@ -54,6 +54,11 @@ internal class SyncGenerationCoordinator(
 ) {
     private val service = SyncGenerationService(repository)
 
+    private companion object {
+        const val ACK_RECOVERY_CAPABILITY = "generation-ack-recovery-v1"
+        const val MAX_COORDINATION_BYTES = 65_536
+    }
+
     private fun phase(
         runId: String?,
         phase: String,
@@ -196,6 +201,45 @@ internal class SyncGenerationCoordinator(
         if (actual != expected) throw SyncGenerationException("invalid $label fields")
     }
 
+    private fun requestSupports(request: JSONObject, capability: String): Boolean {
+        val capabilities = request.optJSONArray("capabilities") ?: return false
+        for (index in 0 until capabilities.length()) {
+            if (capabilities.optString(index) == capability) return true
+        }
+        return false
+    }
+
+    private fun recoveryAcknowledgements(
+        desktopPeerId: String,
+        androidPeerId: String,
+    ): JSONArray =
+        repository.inSyncGenerationTransaction { db ->
+            val acknowledgements = JSONArray()
+            db.rawQuery(
+                    "SELECT ack_json FROM sync_consumed_generations WHERE producer_peer_id=? " +
+                        "AND consumer_peer_id=? AND result IN('consumed','rejected') " +
+                        "ORDER BY consumed_at DESC,generation_id DESC LIMIT 32",
+                    arrayOf(desktopPeerId, androidPeerId),
+                )
+                .use { cursor ->
+                    while (cursor.moveToNext()) {
+                        val acknowledgement = JSONObject(cursor.getString(0))
+                        /* ACKs written before cross-runtime rejection diagnostics
+                         * were normalized can hash '/' differently in Android
+                         * and Python JSON canonicalizers. They remain immutable
+                         * evidence, but must not block recovery of newer,
+                         * compatible ACKs. Consumed ACKs are unaffected. */
+                        if (
+                            acknowledgement.optString("result") != "rejected" ||
+                                !acknowledgement.optString("diagnostic").contains("/")
+                        ) {
+                            acknowledgements.put(acknowledgement)
+                        }
+                    }
+                }
+            acknowledgements
+        }
+
     private fun reconcileArchiveAcknowledgements(
         raw: ByteArray,
         runId: String,
@@ -299,6 +343,7 @@ internal class SyncGenerationCoordinator(
                             "mobile-history-v4",
                             "execution-draft-v1",
                             "generation-archive-v1",
+                            ACK_RECOVERY_CAPABILITY,
                             "session-preparations-v2",
                             "programs-v1",
                         )
@@ -433,6 +478,43 @@ internal class SyncGenerationCoordinator(
                 request.getString("desktop_peer_id"),
             )
             phase(runId, "archive_ack_observed", "desktop-archive-acknowledgements-v1.json")
+            if (requestSupports(request, ACK_RECOVERY_CAPABILITY)) {
+                /* WHY: Android may commit consumption after the desktop
+                 * producer has already published its immutable generation, then
+                 * transport can fail before the ACK reaches that producer.
+                 * CONTRACT: return only exact durable consumer ACK evidence,
+                 * bounded to the newest 32 rows, before either side captures a
+                 * new generation. INVARIANT: recovery never fabricates lineage
+                 * and never rewrites Android consumption history. */
+                val recoveryPath =
+                    File(directory, "android-archive-acknowledgements-v1.json")
+                val recoveryEnvelope =
+                    JSONObject()
+                        .put("format", "trainlog-sync-archive-acknowledgements")
+                        .put("version", 1)
+                        .put("run_id", runId)
+                        .put("android_peer_id", peer)
+                        .put("desktop_peer_id", request.getString("desktop_peer_id"))
+                        .put(
+                            "acknowledgements",
+                            recoveryAcknowledgements(
+                                request.getString("desktop_peer_id"),
+                                peer,
+                            ),
+                        )
+                        .toString()
+                if (recoveryEnvelope.toByteArray(Charsets.UTF_8).size > MAX_COORDINATION_BYTES)
+                    throw SyncGenerationException(
+                        "Android archive acknowledgement envelope exceeds bound"
+                    )
+                publish(recoveryPath, recoveryEnvelope)
+                visibility.confirm(listOf(recoveryPath))
+                phase(
+                    runId,
+                    "android_archive_ack_published",
+                    recoveryPath.name,
+                )
+            }
             /* WHY: transport can fail after immutable generation publication.
              * CONTRACT: an explicit retry for the same run republishes that
              * exact generation and resumes ACK handling; it never recaptures

android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt

@@ -56,6 +56,7 @@ class SyncDeploymentConversationTest {
         val databaseName = "deployment-peer-${UUID.randomUUID()}.db"
         var repository = TrainlogRepository(context, databaseName)
         var localWorker: Process? = null
+        var lostDesktopAckGeneration: String? = null
         try {
             transport.mkdirs()
             // A previous interrupted exchange leaves durable coordination
@@ -180,6 +181,7 @@ class SyncDeploymentConversationTest {
                             "full_generation_trigger_published",
                             "generation_request_observed",
                             "archive_ack_observed",
+                            "android_archive_ack_published",
                             "generation_capture_started",
                             "generation_captured",
                             "generation_objects_published",
@@ -215,6 +217,71 @@ class SyncDeploymentConversationTest {
                         }
                     }
                 }
+                if (suppliedTransport == null && it == 11) {
+                    /* Reproduce the inverse deployed failure: Android has
+                     * durably consumed the newest desktop generation, but its
+                     * ACK was lost before the desktop producer committed it.
+                     * The next conversation must recover that exact Android
+                     * evidence before capturing another desktop generation. */
+                    SQLiteDatabase.openDatabase(
+                        desktop.absolutePath,
+                        null,
+                        SQLiteDatabase.OPEN_READWRITE,
+                    ).use { database ->
+                        val generation =
+                            database.rawQuery(
+                                "SELECT generation_id FROM sync_generations " +
+                                    "WHERE status='acknowledged' ORDER BY generated_at DESC LIMIT 1",
+                                null,
+                            ).use { cursor ->
+                                assertTrue(cursor.moveToFirst())
+                                cursor.getString(0)
+                            }
+                        database.delete(
+                            "sync_acknowledgements",
+                            "generation_id=?",
+                            arrayOf(generation),
+                        )
+                        database.execSQL(
+                            "UPDATE sync_generations SET status='waiting_acknowledgement'," +
+                                "acknowledged_at=NULL WHERE generation_id=?",
+                            arrayOf(generation),
+                        )
+                        database.rawQuery(
+                            "SELECT status FROM sync_generations WHERE generation_id=?",
+                            arrayOf(generation),
+                        ).use { cursor ->
+                            assertTrue(cursor.moveToFirst())
+                            assertEquals("waiting_acknowledgement", cursor.getString(0))
+                        }
+                        lostDesktopAckGeneration = generation
+                    }
+                }
+                if (suppliedTransport == null && it == 12) {
+                    val recovered = checkNotNull(lostDesktopAckGeneration)
+                    SQLiteDatabase.openDatabase(
+                        desktop.absolutePath,
+                        null,
+                        SQLiteDatabase.OPEN_READONLY,
+                    ).use { database ->
+                        database.rawQuery(
+                            "SELECT status FROM sync_generations WHERE generation_id=?",
+                            arrayOf(recovered),
+                        ).use { cursor ->
+                            assertTrue(cursor.moveToFirst())
+                            assertEquals("acknowledged", cursor.getString(0))
+                        }
+                        database.rawQuery(
+                            "SELECT parent_generation_id,status FROM sync_generations " +
+                                "ORDER BY generated_at DESC LIMIT 1",
+                            null,
+                        ).use { cursor ->
+                            assertTrue(cursor.moveToFirst())
+                            assertEquals(recovered, cursor.getString(0))
+                            assertEquals("acknowledged", cursor.getString(1))
+                        }
+                    }
+                }
                 localWorker = null
             }
             repository.close()

tests/test_sync_deployment_tools.py

@@ -78,6 +78,7 @@ class DeploymentToolsTest(unittest.TestCase):
             self.assertIn("heart-rate-v1", inventory["protocols"])
             self.assertIn("session-timeline-v1", inventory["protocols"])
             self.assertIn("bluetooth-files-v1", inventory["protocols"])
+            self.assertIn("generation-ack-recovery-v1", inventory["protocols"])
             self.assertTrue((output / "bin/trainlog").is_file())
             self.assertTrue((output / "bin/trainlog-sync-once").is_file())
             self.assertTrue((output / "bin/trainlog-syncd").is_file())

tools/package_sync_candidate.py

@@ -129,6 +129,7 @@ def main() -> int:
             "generation-manifest-v1",
             "generation-ack-v1",
             "generation-archive-v1",
+            "generation-ack-recovery-v1",
             "bluetooth-files-v1",
             "session-preparations-v2",
             "ai-session-drafts-v2",

tools/sync_generation_exchange.py

@@ -791,8 +791,7 @@ def validate_ack(raw: bytes, pending: sqlite3.Row | tuple) -> dict:
     return value
 
 
-def accept_ack(database: Path, path: Path) -> str:
+def accept_ack_bytes(database: Path, raw: bytes) -> str:
-    raw = path.read_bytes()
     with closing(connect_database(database)) as db:
         require_schema(db); db.execute("BEGIN IMMEDIATE")
         value = strict_json(raw, MAX_MANIFEST)
@@ -862,6 +861,10 @@ def accept_ack(database: Path, path: Path) -> str:
         db.commit(); return status
 
 
+def accept_ack(database: Path, path: Path) -> str:
+    return accept_ack_bytes(database, path.read_bytes())
+
+
 def main() -> None:
     parser = argparse.ArgumentParser()
     sub = parser.add_subparsers(dest="mode", required=True)

tools/sync_peer_worker.py

@@ -35,6 +35,8 @@ CAPS = {
     "session-preparations-v2",
 }
 
+ACK_RECOVERY_CAP = "generation-ack-recovery-v1"
+
 MTP_OPERATION_TIMEOUT_SECONDS = 30.0
 MTP_POLL_INTERVAL_SECONDS = 3.0
 
@@ -115,6 +117,67 @@ def recovery_acknowledgements(db, producer_peer_id: str, consumer_peer_id: str)
     ]
 
 
+def recovery_acknowledgement_envelope(
+    raw: bytes,
+    run_id: str,
+    android_peer_id: str,
+    desktop_peer_id: str,
+) -> list[dict]:
+    """Validate the Android consumer's bounded ACK recovery envelope."""
+    value = generation.strict_json(raw, 65536)
+    expected = {
+        "format",
+        "version",
+        "run_id",
+        "android_peer_id",
+        "desktop_peer_id",
+        "acknowledgements",
+    }
+    if (
+        not isinstance(value, dict)
+        or set(value) != expected
+        or value.get("format") != "trainlog-sync-archive-acknowledgements"
+        or type(value.get("version")) is not int
+        or value.get("version") != 1
+        or value.get("run_id") != run_id
+        or value.get("android_peer_id") != android_peer_id
+        or value.get("desktop_peer_id") != desktop_peer_id
+    ):
+        raise RuntimeError("Android archive acknowledgement envelope is not correlated")
+    acknowledgements = value.get("acknowledgements")
+    if not isinstance(acknowledgements, list) or len(acknowledgements) > 32:
+        raise RuntimeError("invalid Android archive acknowledgement count")
+    if not all(isinstance(ack, dict) for ack in acknowledgements):
+        raise RuntimeError("invalid Android archive acknowledgement")
+    return acknowledgements
+
+
+def reconcile_recovery_acknowledgements(
+    database: Path,
+    raw: bytes,
+    run_id: str,
+    android_peer_id: str,
+    desktop_peer_id: str,
+) -> int:
+    """Repair producer state only from exact ACK evidence retained by Android."""
+    acknowledgements = recovery_acknowledgement_envelope(
+        raw,
+        run_id,
+        android_peer_id,
+        desktop_peer_id,
+    )
+    reconciled = 0
+    for acknowledgement in acknowledgements:
+        result = generation.accept_ack_bytes(
+            database,
+            generation.canonical(acknowledgement),
+        )
+        if result not in ("acknowledged", "rejected", "unchanged"):
+            raise RuntimeError("unexpected recovered ACK result")
+        reconciled += 1
+    return reconciled
+
+
 def run_adapter(
     adapter: Path,
     operation: str,
@@ -630,7 +693,7 @@ def main() -> int:
         "run_id": args.run_id,
         "desktop_peer_id": desktop_peer,
         "android_peer_id": peer["peer_id"],
-        "capabilities": sorted(CAPS),
+        "capabilities": sorted(CAPS | {ACK_RECOVERY_CAP}),
     }
     request_path = args.transport_root / "request-v1.json"
     publish_json(request_path, request)
@@ -678,12 +741,36 @@ def main() -> int:
         producer_peer_id=peer["peer_id"],
         consumer_peer_id=desktop_peer,
     )
-    inbound_ref = args.transport_root / "android-generation-v1.json"
+
     def pump_inbound():
         if pump is not None:
             pump()
         correlated_peer_error(args.transport_root, args.run_id)
 
+    if ACK_RECOVERY_CAP in set(peer.get("capabilities", [])):
+        android_archive_ack = args.transport_root / "android-archive-acknowledgements-v1.json"
+        wait_json(
+            android_archive_ack,
+            deadline,
+            lambda value: value.get("run_id") == args.run_id
+            and value.get("android_peer_id") == peer["peer_id"]
+            and value.get("desktop_peer_id") == desktop_peer,
+            pump_inbound,
+        )
+        recovered = reconcile_recovery_acknowledgements(
+            args.database,
+            bounded(android_archive_ack),
+            args.run_id,
+            peer["peer_id"],
+            desktop_peer,
+        )
+        emit(
+            args.run_id,
+            "android_archive_ack_observed",
+            recovered_acknowledgements=recovered,
+        )
+
+    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
     )