feat(sync): separate full-generation Android trigger
This commit is contained in:
parent
eb6d70fb62
commit
ab3986200e
10 changed files with 252 additions and 64 deletions
|
|
@ -39,6 +39,7 @@ internal fun canonicalExchangeDirectory(): File =
|
||||||
internal fun exchangeArtifactIdentity(displayName: String): String = when (displayName) {
|
internal fun exchangeArtifactIdentity(displayName: String): String = when (displayName) {
|
||||||
"trainlog-exercise-body-zones-v1.json" -> "BODY_ZONES"
|
"trainlog-exercise-body-zones-v1.json" -> "BODY_ZONES"
|
||||||
"trainlog-sync-request-v1.json" -> "SYNC_REQUEST"
|
"trainlog-sync-request-v1.json" -> "SYNC_REQUEST"
|
||||||
|
"trainlog-sync-full-generation-request-v1.json" -> "FULL_GENERATION_SYNC_REQUEST"
|
||||||
"trainlog-mobile-export-v3.json" -> "MOBILE_EXPORT"
|
"trainlog-mobile-export-v3.json" -> "MOBILE_EXPORT"
|
||||||
"trainlog-mobile-equipment-definitions-v1.json" -> "EQUIPMENT_DEFINITIONS"
|
"trainlog-mobile-equipment-definitions-v1.json" -> "EQUIPMENT_DEFINITIONS"
|
||||||
"trainlog-equipment-associations-v2.json" -> "EQUIPMENT_ASSOCIATIONS"
|
"trainlog-equipment-associations-v2.json" -> "EQUIPMENT_ASSOCIATIONS"
|
||||||
|
|
|
||||||
|
|
@ -382,9 +382,9 @@ internal class SyncGenerationCoordinator(
|
||||||
}
|
}
|
||||||
var currentRunId: String? = null
|
var currentRunId: String? = null
|
||||||
return try {
|
return try {
|
||||||
val baselineRunId = request(directory)?.optString("run_id")
|
|
||||||
val peer = publishPeer(directory)
|
val peer = publishPeer(directory)
|
||||||
phase(null, "peer_published", "android-peer-v1.json")
|
phase(null, "peer_published", "android-peer-v1.json")
|
||||||
|
val baselineRunId = request(directory)?.optString("run_id")
|
||||||
/* WHY: trainlog-syncd is intentionally driven by the established
|
/* WHY: trainlog-syncd is intentionally driven by the established
|
||||||
* Android request artifact, not by polling generation internals.
|
* Android request artifact, not by polling generation internals.
|
||||||
* CONTRACT: each explicit foreground attempt publishes the peer
|
* CONTRACT: each explicit foreground attempt publishes the peer
|
||||||
|
|
@ -393,7 +393,11 @@ internal class SyncGenerationCoordinator(
|
||||||
* ACK evidence remain immutable; retry creates new correlation. */
|
* ACK evidence remain immutable; retry creates new correlation. */
|
||||||
afterPeerPublication?.let {
|
afterPeerPublication?.let {
|
||||||
it.invoke()
|
it.invoke()
|
||||||
phase(null, "legacy_trigger_published", "trainlog-sync-request-v1.json")
|
phase(
|
||||||
|
null,
|
||||||
|
"full_generation_trigger_published",
|
||||||
|
FULL_GENERATION_REQUEST_NAME,
|
||||||
|
)
|
||||||
}
|
}
|
||||||
val deadline = System.nanoTime() + timeout.toNanos()
|
val deadline = System.nanoTime() + timeout.toNanos()
|
||||||
val requestRaw =
|
val requestRaw =
|
||||||
|
|
|
||||||
|
|
@ -13,6 +13,23 @@ import java.util.UUID
|
||||||
import kotlinx.coroutines.Dispatchers
|
import kotlinx.coroutines.Dispatchers
|
||||||
import kotlinx.coroutines.withContext
|
import kotlinx.coroutines.withContext
|
||||||
|
|
||||||
|
internal const val FULL_GENERATION_REQUEST_NAME =
|
||||||
|
"trainlog-sync-full-generation-request-v1.json"
|
||||||
|
internal const val LEGACY_SYNC_REQUEST_NAME = "trainlog-sync-request-v1.json"
|
||||||
|
|
||||||
|
internal data class SyncRequestIntent(
|
||||||
|
val requestId: String,
|
||||||
|
val requestedAt: String,
|
||||||
|
) {
|
||||||
|
fun payload(): String =
|
||||||
|
JSONObject()
|
||||||
|
.put("format", "trainlog-sync-request")
|
||||||
|
.put("version", 1)
|
||||||
|
.put("request_id", requestId)
|
||||||
|
.put("requested_at", requestedAt)
|
||||||
|
.toString()
|
||||||
|
}
|
||||||
|
|
||||||
sealed interface SyncRequestResult {
|
sealed interface SyncRequestResult {
|
||||||
/** The full-generation coordinator completed both durable peer ACKs. */
|
/** The full-generation coordinator completed both durable peer ACKs. */
|
||||||
data class Completed(
|
data class Completed(
|
||||||
|
|
@ -60,39 +77,39 @@ class SyncRequestOutbox private constructor(
|
||||||
return SyncRequestResult.Unsupported
|
return SyncRequestResult.Unsupported
|
||||||
}
|
}
|
||||||
|
|
||||||
val requestId =
|
val intent = newIntent()
|
||||||
"sr_" +
|
return publishLegacy(snapshot, intent)
|
||||||
UUID.randomUUID()
|
}
|
||||||
.toString()
|
|
||||||
|
|
||||||
val payload =
|
internal fun newIntent(): SyncRequestIntent =
|
||||||
JSONObject()
|
SyncRequestIntent(
|
||||||
.put(
|
requestId = "sr_${UUID.randomUUID()}",
|
||||||
"format",
|
requestedAt = OffsetDateTime.now().toString(),
|
||||||
"trainlog-sync-request",
|
)
|
||||||
)
|
|
||||||
.put(
|
|
||||||
"version",
|
|
||||||
1,
|
|
||||||
)
|
|
||||||
.put(
|
|
||||||
"request_id",
|
|
||||||
requestId,
|
|
||||||
)
|
|
||||||
.put(
|
|
||||||
"requested_at",
|
|
||||||
OffsetDateTime.now()
|
|
||||||
.toString(),
|
|
||||||
)
|
|
||||||
.toString()
|
|
||||||
|
|
||||||
val displayName =
|
internal fun publishFullGeneration(
|
||||||
"trainlog-sync-request-v1.json"
|
snapshot: DirectExchangeSnapshot,
|
||||||
val error = snapshot.writeJson(displayName, payload)
|
intent: SyncRequestIntent,
|
||||||
|
): SyncRequestResult = publish(snapshot, FULL_GENERATION_REQUEST_NAME, intent)
|
||||||
|
|
||||||
|
internal fun publishLegacy(
|
||||||
|
snapshot: DirectExchangeSnapshot,
|
||||||
|
intent: SyncRequestIntent,
|
||||||
|
): SyncRequestResult = publish(snapshot, LEGACY_SYNC_REQUEST_NAME, intent)
|
||||||
|
|
||||||
|
private fun publish(
|
||||||
|
snapshot: DirectExchangeSnapshot,
|
||||||
|
displayName: String,
|
||||||
|
intent: SyncRequestIntent,
|
||||||
|
): SyncRequestResult {
|
||||||
|
if (Build.VERSION.SDK_INT < Build.VERSION_CODES.Q) {
|
||||||
|
return SyncRequestResult.Unsupported
|
||||||
|
}
|
||||||
|
val error = snapshot.writeJson(displayName, intent.payload())
|
||||||
return if (error == null) {
|
return if (error == null) {
|
||||||
try {
|
try {
|
||||||
visibility.confirm(listOf(java.io.File(canonicalExchangeDirectory(), displayName)))
|
visibility.confirm(listOf(java.io.File(canonicalExchangeDirectory(), displayName)))
|
||||||
SyncRequestResult.Requested(requestId)
|
SyncRequestResult.Requested(intent.requestId)
|
||||||
} catch (error: Exception) {
|
} catch (error: Exception) {
|
||||||
SyncRequestResult.Error(error.message ?: "Publication MTP impossible.")
|
SyncRequestResult.Error(error.message ?: "Publication MTP impossible.")
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -60,29 +60,29 @@ internal suspend fun publishBundleAndRequest(
|
||||||
terminalLedger: BackgroundSyncRunLedger? = null,
|
terminalLedger: BackgroundSyncRunLedger? = null,
|
||||||
): SyncRequestResult = withContext(Dispatchers.IO) {
|
): SyncRequestResult = withContext(Dispatchers.IO) {
|
||||||
try {
|
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 directory = canonicalExchangeDirectory()
|
||||||
val coordinator = SyncGenerationForegroundCoordinator(repository, visibility)
|
val coordinator = SyncGenerationForegroundCoordinator(repository, visibility)
|
||||||
return@withContext when (
|
return@withContext when (
|
||||||
val result = coordinator.run(
|
val result = coordinator.run(
|
||||||
directory,
|
directory,
|
||||||
afterPeerPublication = {
|
afterPeerPublication = {
|
||||||
when (val request = requestOutbox.requestSync(snapshot)) {
|
val opened = exporter.openStorageSnapshot()
|
||||||
|
if (opened is com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Error) {
|
||||||
|
throw IllegalStateException(opened.message)
|
||||||
|
}
|
||||||
|
val snapshot =
|
||||||
|
(opened as com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Ready)
|
||||||
|
.snapshot
|
||||||
|
val requestIntent = requestOutbox.newIntent()
|
||||||
|
/* WHY: a strict V3 snapshot cannot represent causal
|
||||||
|
* tombstones, but the complete generation can. CONTRACT:
|
||||||
|
* publish and confirm the modern trigger before attempting
|
||||||
|
* optional legacy compatibility. INVARIANT: an old desktop
|
||||||
|
* sees this intent only after a fresh V3 bundle succeeds. */
|
||||||
|
when (
|
||||||
|
val request =
|
||||||
|
requestOutbox.publishFullGeneration(snapshot, requestIntent)
|
||||||
|
) {
|
||||||
is SyncRequestResult.Requested -> Unit
|
is SyncRequestResult.Requested -> Unit
|
||||||
SyncRequestResult.Unsupported ->
|
SyncRequestResult.Unsupported ->
|
||||||
throw IllegalStateException("Generation synchronization is unsupported.")
|
throw IllegalStateException("Generation synchronization is unsupported.")
|
||||||
|
|
@ -90,6 +90,27 @@ internal suspend fun publishBundleAndRequest(
|
||||||
is SyncRequestResult.Completed ->
|
is SyncRequestResult.Completed ->
|
||||||
throw IllegalStateException("Unexpected completed request signal.")
|
throw IllegalStateException("Unexpected completed request signal.")
|
||||||
}
|
}
|
||||||
|
when (exporter.exportMobileBundle(snapshot)) {
|
||||||
|
is SyncExportResult.Exported ->
|
||||||
|
when (requestOutbox.publishLegacy(snapshot, requestIntent)) {
|
||||||
|
is SyncRequestResult.Requested ->
|
||||||
|
logDirectExchange(
|
||||||
|
"trainlog-sync-request-v1.json",
|
||||||
|
"legacy_compatibility_published",
|
||||||
|
)
|
||||||
|
else ->
|
||||||
|
logDirectExchange(
|
||||||
|
"trainlog-sync-request-v1.json",
|
||||||
|
"legacy_compatibility_unavailable",
|
||||||
|
)
|
||||||
|
}
|
||||||
|
SyncExportResult.Unsupported,
|
||||||
|
is SyncExportResult.Error ->
|
||||||
|
logDirectExchange(
|
||||||
|
"trainlog-sync-request-v1.json",
|
||||||
|
"legacy_compatibility_unavailable",
|
||||||
|
)
|
||||||
|
}
|
||||||
},
|
},
|
||||||
intent = SyncConversationIntent.NEW_EXPLICIT,
|
intent = SyncConversationIntent.NEW_EXPLICIT,
|
||||||
explicitClaim = explicitClaim,
|
explicitClaim = explicitClaim,
|
||||||
|
|
|
||||||
|
|
@ -177,7 +177,7 @@ class SyncDeploymentConversationTest {
|
||||||
tracePhases.containsAll(
|
tracePhases.containsAll(
|
||||||
listOf(
|
listOf(
|
||||||
"peer_published",
|
"peer_published",
|
||||||
"legacy_trigger_published",
|
"full_generation_trigger_published",
|
||||||
"generation_request_observed",
|
"generation_request_observed",
|
||||||
"archive_ack_observed",
|
"archive_ack_observed",
|
||||||
"generation_capture_started",
|
"generation_capture_started",
|
||||||
|
|
|
||||||
|
|
@ -3,6 +3,7 @@ package com.labfytools.trainlog.data
|
||||||
import java.nio.file.Files
|
import java.nio.file.Files
|
||||||
import org.json.JSONObject
|
import org.json.JSONObject
|
||||||
import org.junit.Assert.assertNotEquals
|
import org.junit.Assert.assertNotEquals
|
||||||
|
import org.junit.Assert.assertEquals
|
||||||
import org.junit.Assert.assertTrue
|
import org.junit.Assert.assertTrue
|
||||||
import org.junit.Test
|
import org.junit.Test
|
||||||
import org.junit.runner.RunWith
|
import org.junit.runner.RunWith
|
||||||
|
|
@ -32,4 +33,27 @@ class SyncRequestOutboxTest {
|
||||||
directory.deleteRecursively()
|
directory.deleteRecursively()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
fun oneIntentUsesSameIdentityAcrossFullGenerationAndLegacyChannels() {
|
||||||
|
val directory = Files.createTempDirectory("trainlog-dual-request-").toFile()
|
||||||
|
try {
|
||||||
|
val outbox = SyncRequestOutbox { directExchangeDirectoryForPath(directory) }
|
||||||
|
val snapshot =
|
||||||
|
(DirectExchangePublisher { directExchangeDirectoryForPath(directory) }.snapshot()
|
||||||
|
as DirectExchangeSnapshotResult.Ready).snapshot
|
||||||
|
val intent = outbox.newIntent()
|
||||||
|
|
||||||
|
assertTrue(outbox.publishFullGeneration(snapshot, intent) is SyncRequestResult.Requested)
|
||||||
|
assertTrue(outbox.publishLegacy(snapshot, intent) is SyncRequestResult.Requested)
|
||||||
|
|
||||||
|
val full = JSONObject(java.io.File(directory, FULL_GENERATION_REQUEST_NAME).readText())
|
||||||
|
val legacy = JSONObject(java.io.File(directory, LEGACY_SYNC_REQUEST_NAME).readText())
|
||||||
|
assertEquals(intent.requestId, full.getString("request_id"))
|
||||||
|
assertEquals(intent.requestId, legacy.getString("request_id"))
|
||||||
|
assertEquals(full.toString(), legacy.toString())
|
||||||
|
} finally {
|
||||||
|
directory.deleteRecursively()
|
||||||
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -126,17 +126,65 @@ class SyncdRoutingTest(unittest.TestCase):
|
||||||
self.assertEqual(1, len(commands))
|
self.assertEqual(1, len(commands))
|
||||||
self.assertEqual("sy_old", json.loads(state.read_text())["run_id"])
|
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):
|
def test_android_production_publishes_full_generation_before_optional_legacy(self):
|
||||||
source = (ROOT / "android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt").read_text()
|
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(
|
production = source.split("internal suspend fun publishBundleAndRequest", 1)[1].split(
|
||||||
"internal suspend fun publishLegacyBundleAndRequest", 1
|
"internal suspend fun publishLegacyBundleAndRequest", 1
|
||||||
)[0]
|
)[0]
|
||||||
self.assertIn(
|
self.assertIn("coordinator.run", production)
|
||||||
"SyncGenerationForegroundCoordinator(repository, visibility).run", production
|
|
||||||
)
|
|
||||||
self.assertNotIn("generation-opt-in", production)
|
self.assertNotIn("generation-opt-in", production)
|
||||||
self.assertIn("exportMobileBundle", production)
|
self.assertIn("exportMobileBundle", production)
|
||||||
self.assertIn("generation below", production)
|
self.assertLess(
|
||||||
|
production.index("publishFullGeneration"), production.index("exportMobileBundle")
|
||||||
|
)
|
||||||
|
self.assertGreater(
|
||||||
|
production.index("publishLegacy"), production.index("exportMobileBundle")
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_full_generation_wins_and_retires_older_legacy(self):
|
||||||
|
with tempfile.TemporaryDirectory() as directory:
|
||||||
|
root = Path(directory)
|
||||||
|
old = "sr_11111111-1111-4111-8111-111111111111"
|
||||||
|
new = "sr_22222222-2222-4222-8222-222222222222"
|
||||||
|
self._write_request(root / syncd.REQUEST_NAME, old, "2026-09-20T11:00:00Z")
|
||||||
|
self._write_request(
|
||||||
|
root / syncd.FULL_GENERATION_REQUEST_NAME, new, "2026-09-20T12:00:00Z"
|
||||||
|
)
|
||||||
|
selected = syncd.select_request(root, [])
|
||||||
|
self.assertEqual((new, "full_generation", [old]), selected)
|
||||||
|
self.assertIsNone(syncd.select_request(root, [old, new]))
|
||||||
|
|
||||||
|
def test_dual_channel_same_intent_is_one_request(self):
|
||||||
|
with tempfile.TemporaryDirectory() as directory:
|
||||||
|
root = Path(directory)
|
||||||
|
request = "sr_22222222-2222-4222-8222-222222222222"
|
||||||
|
for name in (syncd.REQUEST_NAME, syncd.FULL_GENERATION_REQUEST_NAME):
|
||||||
|
self._write_request(root / name, request, "2026-09-20T12:00:00Z")
|
||||||
|
self.assertEqual(
|
||||||
|
(request, "full_generation", []), syncd.select_request(root, [])
|
||||||
|
)
|
||||||
|
self.assertIsNone(syncd.select_request(root, [request]))
|
||||||
|
|
||||||
|
def test_newer_legacy_remains_eligible_after_full_generation(self):
|
||||||
|
with tempfile.TemporaryDirectory() as directory:
|
||||||
|
root = Path(directory)
|
||||||
|
full = "sr_22222222-2222-4222-8222-222222222222"
|
||||||
|
legacy = "sr_33333333-3333-4333-8333-333333333333"
|
||||||
|
self._write_request(
|
||||||
|
root / syncd.FULL_GENERATION_REQUEST_NAME, full, "2026-09-20T12:00:00Z"
|
||||||
|
)
|
||||||
|
self._write_request(root / syncd.REQUEST_NAME, legacy, "2026-09-20T13:00:00Z")
|
||||||
|
self.assertEqual((full, "full_generation", []), syncd.select_request(root, []))
|
||||||
|
self.assertEqual((legacy, "legacy", [full]), syncd.select_request(root, [full]))
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _write_request(path, request_id, requested_at):
|
||||||
|
path.write_text(json.dumps({
|
||||||
|
"format": "trainlog-sync-request",
|
||||||
|
"version": 1,
|
||||||
|
"request_id": request_id,
|
||||||
|
"requested_at": requested_at,
|
||||||
|
}))
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
|
|
|
||||||
|
|
@ -19,6 +19,8 @@ from pathlib import Path
|
||||||
|
|
||||||
STOP = False
|
STOP = False
|
||||||
REQUEST_NAME = "trainlog-sync-request-v1.json"
|
REQUEST_NAME = "trainlog-sync-request-v1.json"
|
||||||
|
FULL_GENERATION_REQUEST_NAME = "trainlog-sync-full-generation-request-v1.json"
|
||||||
|
SEEN_REQUEST_LIMIT = 64
|
||||||
REQUEST_ID = re.compile(r"sr_[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}")
|
REQUEST_ID = re.compile(r"sr_[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}")
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -90,7 +92,7 @@ def load_generation_config(path: Path) -> dict:
|
||||||
return value
|
return value
|
||||||
|
|
||||||
|
|
||||||
def request_id(path: Path) -> str | None:
|
def request(path: Path) -> tuple[str, datetime] | None:
|
||||||
try:
|
try:
|
||||||
value = json.loads(path.read_text(encoding="utf-8"))
|
value = json.loads(path.read_text(encoding="utf-8"))
|
||||||
except (OSError, UnicodeError, json.JSONDecodeError):
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
||||||
|
|
@ -106,7 +108,58 @@ def request_id(path: Path) -> str | None:
|
||||||
and isinstance(candidate, str)
|
and isinstance(candidate, str)
|
||||||
and REQUEST_ID.fullmatch(candidate)
|
and REQUEST_ID.fullmatch(candidate)
|
||||||
):
|
):
|
||||||
return candidate
|
requested_at = value.get("requested_at")
|
||||||
|
if not isinstance(requested_at, str):
|
||||||
|
return None
|
||||||
|
try:
|
||||||
|
timestamp = datetime.fromisoformat(requested_at.replace("Z", "+00:00"))
|
||||||
|
except ValueError:
|
||||||
|
return None
|
||||||
|
if timestamp.tzinfo is None:
|
||||||
|
return None
|
||||||
|
return candidate, timestamp
|
||||||
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
def load_request_state(path: Path) -> tuple[dict, list[str]]:
|
||||||
|
try:
|
||||||
|
state = json.loads(path.read_text(encoding="utf-8"))
|
||||||
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
||||||
|
return {}, []
|
||||||
|
if not isinstance(state, dict):
|
||||||
|
return {}, []
|
||||||
|
seen = [item for item in state.get("seen_request_ids", [])
|
||||||
|
if isinstance(item, str) and REQUEST_ID.fullmatch(item)]
|
||||||
|
historical = state.get("request_id")
|
||||||
|
if isinstance(historical, str) and REQUEST_ID.fullmatch(historical) and historical not in seen:
|
||||||
|
seen.append(historical)
|
||||||
|
return state, seen[-SEEN_REQUEST_LIMIT:]
|
||||||
|
|
||||||
|
|
||||||
|
def select_request(root: Path, seen: list[str]) -> tuple[str, str, list[str]] | None:
|
||||||
|
"""Select one unseen intent while preventing an older cross-channel replay.
|
||||||
|
|
||||||
|
WHY: modern Android may publish the same intent on full-generation and
|
||||||
|
legacy channels, while stale legacy files remain durable on MTP. CONTRACT:
|
||||||
|
full-generation wins and an older alternate intent is retired. INVARIANT:
|
||||||
|
one request_id starts at most one daemon conversation.
|
||||||
|
"""
|
||||||
|
full = request(root / FULL_GENERATION_REQUEST_NAME)
|
||||||
|
legacy = request(root / REQUEST_NAME)
|
||||||
|
unseen_full = full if full is not None and full[0] not in seen else None
|
||||||
|
unseen_legacy = legacy if legacy is not None and legacy[0] not in seen else None
|
||||||
|
if unseen_full is not None:
|
||||||
|
selected_id, selected_at = unseen_full
|
||||||
|
retired = list(seen)
|
||||||
|
if (
|
||||||
|
unseen_legacy is not None
|
||||||
|
and unseen_legacy[0] != selected_id
|
||||||
|
and unseen_legacy[1] <= selected_at
|
||||||
|
):
|
||||||
|
retired.append(unseen_legacy[0])
|
||||||
|
return selected_id, "full_generation", retired
|
||||||
|
if unseen_legacy is not None:
|
||||||
|
return unseen_legacy[0], "legacy", list(seen)
|
||||||
return None
|
return None
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -123,16 +176,13 @@ def run_full_generation(args: argparse.Namespace, config_path: Path) -> int:
|
||||||
)
|
)
|
||||||
if pull.returncode != 0:
|
if pull.returncode != 0:
|
||||||
return 3
|
return 3
|
||||||
correlated_request = request_id(root / REQUEST_NAME)
|
_, seen = load_request_state(args.state)
|
||||||
if correlated_request is None:
|
selected = select_request(root, seen)
|
||||||
|
if selected is None:
|
||||||
return 3
|
return 3
|
||||||
if args.state.is_file():
|
correlated_request, request_channel, seen = selected
|
||||||
try:
|
seen.append(correlated_request)
|
||||||
previous = json.loads(args.state.read_text(encoding="utf-8"))
|
seen = list(dict.fromkeys(seen))[-SEEN_REQUEST_LIMIT:]
|
||||||
if previous.get("request_id") == correlated_request:
|
|
||||||
return 3
|
|
||||||
except (OSError, UnicodeError, json.JSONDecodeError):
|
|
||||||
pass
|
|
||||||
run_id = "sy_" + str(uuid.uuid4())
|
run_id = "sy_" + str(uuid.uuid4())
|
||||||
now = datetime.now(timezone.utc).isoformat()
|
now = datetime.now(timezone.utc).isoformat()
|
||||||
atomic_json(
|
atomic_json(
|
||||||
|
|
@ -142,6 +192,8 @@ def run_full_generation(args: argparse.Namespace, config_path: Path) -> int:
|
||||||
"version": 1,
|
"version": 1,
|
||||||
"run_id": run_id,
|
"run_id": run_id,
|
||||||
"request_id": correlated_request,
|
"request_id": correlated_request,
|
||||||
|
"request_channel": request_channel,
|
||||||
|
"seen_request_ids": seen,
|
||||||
"trigger": args.trigger,
|
"trigger": args.trigger,
|
||||||
"requested_mode": "full_generation_v1",
|
"requested_mode": "full_generation_v1",
|
||||||
"effective_mode": "full_generation_v1",
|
"effective_mode": "full_generation_v1",
|
||||||
|
|
|
||||||
|
|
@ -602,6 +602,18 @@ TrainlogStatus trainlog_generation_mtp_pull(const TrainlogGenerationMtpIo *io,
|
||||||
status = pull_named_file(
|
status = pull_named_file(
|
||||||
io, &selection, selection.root, "trainlog-sync-request-v1.json", local, false);
|
io, &selection, selection.root, "trainlog-sync-request-v1.json", local, false);
|
||||||
}
|
}
|
||||||
|
if (status == TRAINLOG_STATUS_OK) {
|
||||||
|
/* WHY: strict legacy V3 may be unavailable while a complete causal
|
||||||
|
* generation remains valid. CONTRACT: pull the dedicated bounded
|
||||||
|
* coordination object without scanning the exchange tree. INVARIANT:
|
||||||
|
* this read never acknowledges, deletes, or rewrites remote state. */
|
||||||
|
status = pull_named_file(io,
|
||||||
|
&selection,
|
||||||
|
selection.root,
|
||||||
|
"trainlog-sync-full-generation-request-v1.json",
|
||||||
|
local,
|
||||||
|
false);
|
||||||
|
}
|
||||||
if (status == TRAINLOG_STATUS_OK) {
|
if (status == TRAINLOG_STATUS_OK) {
|
||||||
status = pull_named_file(
|
status = pull_named_file(
|
||||||
io, &selection, selection.root, "android-generation-v1.json", local, false);
|
io, &selection, selection.root, "android-generation-v1.json", local, false);
|
||||||
|
|
|
||||||
|
|
@ -244,6 +244,12 @@ static int test_pull_and_manifest_last(void) {
|
||||||
"trainlog-sync-request-v1.json",
|
"trainlog-sync-request-v1.json",
|
||||||
"{\"format\":\"trainlog-sync-request\",\"version\":1,\"request_id\":\"sr_11111111-"
|
"{\"format\":\"trainlog-sync-request\",\"version\":1,\"request_id\":\"sr_11111111-"
|
||||||
"1111-4111-8111-111111111111\",\"requested_at\":\"2026-09-20T12:00:00Z\"}");
|
"1111-4111-8111-111111111111\",\"requested_at\":\"2026-09-20T12:00:00Z\"}");
|
||||||
|
add(14,
|
||||||
|
2,
|
||||||
|
false,
|
||||||
|
"trainlog-sync-full-generation-request-v1.json",
|
||||||
|
"{\"format\":\"trainlog-sync-request\",\"version\":1,\"request_id\":\"sr_11111111-"
|
||||||
|
"1111-4111-8111-111111111111\",\"requested_at\":\"2026-09-20T12:00:00Z\"}");
|
||||||
add(5, 2, true, "android-objects", NULL);
|
add(5, 2, true, "android-objects", NULL);
|
||||||
add(6, 5, true, "generations", NULL);
|
add(6, 5, true, "generations", NULL);
|
||||||
add(7, 6, true, "gen_11111111-1111-4111-8111-111111111111", NULL);
|
add(7, 6, true, "gen_11111111-1111-4111-8111-111111111111", NULL);
|
||||||
|
|
@ -261,6 +267,9 @@ static int test_pull_and_manifest_last(void) {
|
||||||
root,
|
root,
|
||||||
"android-objects/generations/gen_11111111-1111-4111-8111-111111111111/payload.json"));
|
"android-objects/generations/gen_11111111-1111-4111-8111-111111111111/payload.json"));
|
||||||
CHECK(access(path, R_OK) == 0);
|
CHECK(access(path, R_OK) == 0);
|
||||||
|
CHECK(trainlog_test_join_path(
|
||||||
|
path, sizeof(path), root, "trainlog-sync-full-generation-request-v1.json"));
|
||||||
|
CHECK(access(path, R_OK) == 0);
|
||||||
CHECK(trainlog_test_join_path(path, sizeof(path), root, "trainlog-sync-request-v1.json"));
|
CHECK(trainlog_test_join_path(path, sizeof(path), root, "trainlog-sync-request-v1.json"));
|
||||||
CHECK(access(path, R_OK) == 0);
|
CHECK(access(path, R_OK) == 0);
|
||||||
CHECK(trainlog_test_join_path(path, sizeof(path), root, "out"));
|
CHECK(trainlog_test_join_path(path, sizeof(path), root, "out"));
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue