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 new file mode 100644 index 0000000..4a0fc58 --- /dev/null +++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationForegroundCoordinator.kt @@ -0,0 +1,61 @@ +package com.labfytools.trainlog.data + +import java.io.File +import java.io.FileOutputStream +import java.time.Duration +import java.util.UUID +import org.json.JSONArray +import org.json.JSONObject + +internal sealed interface ForegroundGenerationResult { + data class Completed(val runId: String) : ForegroundGenerationResult + data class Error(val message: String) : ForegroundGenerationResult +} + +/** Owns the bounded, user-initiated Android side of a generation conversation. */ +internal class SyncGenerationForegroundCoordinator(private val repository: TrainlogRepository) { + private val service = SyncGenerationService(repository) + + 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() + } + 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}") + 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") + } +} diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationService.kt b/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationService.kt index 82dbd17..e39cb71 100644 --- a/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationService.kt +++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationService.kt @@ -105,6 +105,10 @@ internal class SyncGenerationService(private val repository: TrainlogRepository) value } + /* CONTRACT: capability advertisements bind to this installation identity; + * callers cannot supply or replace it. */ + internal fun peerId(): String = stablePeerId("android") + fun capture( ownedRoot: File, consumerPeerId: String, diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/TrainlogRepository.kt b/android/app/src/main/java/com/labfytools/trainlog/data/TrainlogRepository.kt index 704e2fd..af1b850 100644 --- a/android/app/src/main/java/com/labfytools/trainlog/data/TrainlogRepository.kt +++ b/android/app/src/main/java/com/labfytools/trainlog/data/TrainlogRepository.kt @@ -4127,6 +4127,12 @@ class TrainlogRepository( } catch (error: Exception) { return ExecutionDraftImportResult.Invalid(error.message ?: "Invalid execution draft.") } + val localPayloads = JSONObject(buildExecutionDraftExportV1Json()).getJSONArray("drafts").let { values -> + buildMap { for (index in 0 until values.length()) { + val value = values.getJSONObject(index) + put(value.getString("session_id"), value) + } } + } val db = database.writableDatabase var active = 0; var pending = 0; var unchanged = 0; var stale = 0 return try { @@ -4147,10 +4153,33 @@ class TrainlogRepository( arrayOf(sessionId, draft.revisionId), ).use { if (!it.moveToFirst()) null else Pair(if (it.isNull(0)) null else it.getString(0), it.getString(1)) } if (known != null) { - if (known != Pair(draft.parentRevisionId, encoded)) throw SyncLifecycleConflict(sessionId) + /* CONTRACT: active/pending is receiver-owned scheduling state, not + * revision content. A peer may echo the same immutable revision + * with a different local state without causing a conflict. */ + val knownPayload = JSONObject(known.second).apply { remove("state") } + val receivedPayload = JSONObject(encoded).apply { remove("state") } + if (known.first != draft.parentRevisionId || + canonicalCausal(knownPayload) != canonicalCausal(receivedPayload)) + throw SyncLifecycleConflict(sessionId) if (current?.first == draft.revisionId) unchanged += 1 else stale += 1 continue } + if (current?.first == draft.revisionId) { + val localPayload = localPayloads[sessionId] ?: throw SyncLifecycleConflict(sessionId) + val localCausal = JSONObject(localPayload.toString()).apply { remove("state") } + val receivedCausal = JSONObject(encoded).apply { remove("state") } + if (canonicalCausal(localCausal) != canonicalCausal(receivedCausal)) + throw SyncLifecycleConflict(sessionId) + /* Persist the locally-created revision on first peer echo so + * later replays retain the same contradiction protection. */ + db.insertOrThrow("execution_draft_revisions", null, ContentValues().apply { + put("session_id", sessionId); put("revision_id", draft.revisionId) + putOptionalString("parent_revision_id", draft.parentRevisionId) + put("payload_json", encoded) + }) + unchanged += 1 + continue + } if (current != null && current.first != draft.parentRevisionId) throw SyncLifecycleConflict(sessionId) if (current?.second == "active") { persistImportedExecutionDraft(db, draft, payload); active += 1 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 85a3b47..741fadd 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 @@ -25,6 +25,10 @@ import com.labfytools.trainlog.data.SyncExportResult import com.labfytools.trainlog.data.SyncReceiptResult 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.SyncGenerationForegroundCoordinator +import com.labfytools.trainlog.data.canonicalExchangeDirectory import com.labfytools.trainlog.data.logDirectExchange import com.labfytools.trainlog.data.directStoragePermissionIntent import com.labfytools.trainlog.ui.theme.LocalTrainlogColors @@ -37,9 +41,24 @@ import com.labfytools.trainlog.R /* Package-visible for orchestration tests: this is the production ordering * boundary, not a duplicate test-only implementation. */ internal suspend fun publishBundleAndRequest( + repository: TrainlogRepository, exporter: SyncExporter, requestOutbox: SyncRequestOutbox, ): SyncRequestResult = withContext(Dispatchers.IO) { + val optIn = java.io.File(canonicalExchangeDirectory(), "trainlog-sync-generation-opt-in-v1.json") + if (optIn.isFile) { + val enabled = try { + val value = org.json.JSONObject(optIn.readText()) + value.keys().asSequence().toSet() == setOf("format", "version", "enabled") && + value.optString("format") == "trainlog-sync-generation-opt-in" && + value.opt("version") is Int && value.getInt("version") == 1 && value.opt("enabled") == true + } catch (_: Exception) { false } + if (!enabled) return@withContext SyncRequestResult.Error("Invalid generation synchronization opt-in.") + return@withContext when (val result = SyncGenerationForegroundCoordinator(repository).run(canonicalExchangeDirectory())) { + is ForegroundGenerationResult.Completed -> SyncRequestResult.Requested(result.runId) + is ForegroundGenerationResult.Error -> SyncRequestResult.Error(result.message) + } + } logDirectExchange("SYNC_BUNDLE", "coordinator.export.begin") when (val opened = exporter.openStorageSnapshot()) { is com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Error -> @@ -78,6 +97,7 @@ internal fun visibleReceiptStatus( @Composable fun SyncScreen( + repository: TrainlogRepository, inbox: SyncCatalogInbox, exporter: SyncExporter, requestOutbox: SyncRequestOutbox, @@ -315,7 +335,7 @@ fun SyncScreen( syncRunning = true coroutineScope.launch { try { - when (val result = publishBundleAndRequest(exporter, requestOutbox)) { + when (val result = publishBundleAndRequest(repository, exporter, requestOutbox)) { is SyncRequestResult.Requested -> { success = true pendingRequestId = result.requestId diff --git a/android/app/src/main/java/com/labfytools/trainlog/ui/TrainlogApp.kt b/android/app/src/main/java/com/labfytools/trainlog/ui/TrainlogApp.kt index f2737f9..28d870c 100644 --- a/android/app/src/main/java/com/labfytools/trainlog/ui/TrainlogApp.kt +++ b/android/app/src/main/java/com/labfytools/trainlog/ui/TrainlogApp.kt @@ -139,7 +139,7 @@ fun TrainlogApp(repository: TrainlogRepository, exporter: SyncExporter, inbox: S AppRoute.Statistics -> StatisticsDashboard(repository) AppRoute.BodyMeasurements -> BodyScreen(repository, bodyState, { exportSnapshot() }, { back() }) AppRoute.LatestMaxima -> LatestMaximaScreen(repository) - AppRoute.Sync -> SyncScreen(inbox, exporter, requestOutbox, { exportSnapshot(); catalogRevision++ }, { back() }) + AppRoute.Sync -> SyncScreen(repository, inbox, exporter, requestOutbox, { exportSnapshot(); catalogRevision++ }, { back() }) AppRoute.Settings -> SettingsScreen(inbox) { exportSnapshot(); catalogRevision++ } }} } diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt new file mode 100644 index 0000000..2442812 --- /dev/null +++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt @@ -0,0 +1,99 @@ +package com.labfytools.trainlog.data + +import android.content.Context +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.SessionSetDraft +import com.labfytools.trainlog.model.TrackingMode +import java.io.File +import java.time.Duration +import java.util.UUID +import org.json.JSONObject +import org.junit.Assert.assertEquals +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 SyncDeploymentConversationTest { + private val context: Context = ApplicationProvider.getApplicationContext() + + private fun awaitFile(file: File): File { + val deadline = System.nanoTime() + Duration.ofSeconds(30).toNanos() + while (!file.isFile) { + if (System.nanoTime() >= deadline) error("timeout waiting for ${file.name}") + Thread.sleep(25) + } + return file + } + + 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() + } + check(temporary.renameTo(file)) { "cannot publish ${file.name}" } + } + + @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 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 coordinator = SyncGenerationForegroundCoordinator(repository) + val release = System.getenv("TRAINLOG_SYNC_TEST_RELEASE_FILE") + val executor = java.util.concurrent.Executors.newSingleThreadExecutor() + val result = executor.submit { + 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() + 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() + } + assertTrue(result.get(35, java.util.concurrent.TimeUnit.SECONDS) is ForegroundGenerationResult.Completed) + executor.shutdownNow() + repository.close() + repository = TrainlogRepository(context, databaseName) + assertTrue(repository.loadActiveSessionDraft() is ActiveDraftLoadResult.Loaded) + localWorker?.let { worker -> + val output = worker.inputStream.bufferedReader().readText() + assertEquals("worker failed:\n$output", 0, worker.waitFor()) + assertTrue(output.contains("\"result\":\"completed\"")) + } + } finally { + localWorker?.destroyForcibly() + repository.close() + context.deleteDatabase(databaseName) + if (suppliedTransport == null) transport.deleteRecursively() + } + } +} diff --git a/android/app/src/test/java/com/labfytools/trainlog/ui/SyncBundlePublicationTest.kt b/android/app/src/test/java/com/labfytools/trainlog/ui/SyncBundlePublicationTest.kt index 3335ff0..8c1a070 100644 --- a/android/app/src/test/java/com/labfytools/trainlog/ui/SyncBundlePublicationTest.kt +++ b/android/app/src/test/java/com/labfytools/trainlog/ui/SyncBundlePublicationTest.kt @@ -100,7 +100,7 @@ class SyncBundlePublicationTest { assertTrue(feedback is SaveFeedbackResult.Saved) val writesBeforeFirstClick = directory.writeCounts.toMap() - assertTrue(publishBundleAndRequest(exporter, outbox) is SyncRequestResult.Requested) + assertTrue(publishBundleAndRequest(repository, exporter, outbox) is SyncRequestResult.Requested) assertEquals(2, directory.enumerationCount) val required = listOf( "trainlog-mobile-export-v3.json", @@ -138,7 +138,7 @@ class SyncBundlePublicationTest { ShadowLog.clear() val bodyCreatesBefore = directory.createCount("trainlog-exercise-body-zones-v1.json") val writesBeforeSecondClick = directory.writeCounts.toMap() - assertTrue(publishBundleAndRequest(exporter, outbox) is SyncRequestResult.Requested) + assertTrue(publishBundleAndRequest(repository, exporter, outbox) is SyncRequestResult.Requested) assertEquals(3, directory.enumerationCount) val phases = ShadowLog.getLogsForTag(DIRECT_EXCHANGE_LOG_TAG).map { it.msg } assertEquals(1, phases.count { it.startsWith("SYNC_BUNDLE coordinator.export.begin") }) diff --git a/tests/test_sync_orchestrator.py b/tests/test_sync_orchestrator.py index 50ec59a..7dcbfee 100644 --- a/tests/test_sync_orchestrator.py +++ b/tests/test_sync_orchestrator.py @@ -1,31 +1,43 @@ -import json, os, subprocess, tempfile, unittest, uuid +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" -FIXTURE=ROOT/"tests/sync_orchestrator_peer_fixture.py" +def identity(prefix): return prefix+str(uuid.uuid4()) class OrchestratorTest(unittest.TestCase): - def test_correlated_worker_report_is_durable_and_reopenable(self): - for trigger in ("web","android"): - with self.subTest(trigger=trigger), tempfile.TemporaryDirectory() as directory: - root=Path(directory);db=root/"trainlog.db";db.touch();state=Path(str(db)+".sync-run.json") - run="sy_"+str(uuid.uuid4());request="sy_"+str(uuid.uuid4()) - state.write_text(json.dumps({"run_id":run,"request_id":request,"trigger":trigger,"phase":"requested","progress_revision":1,"result":"running"})) - config=root/"config.json";config.write_text(json.dumps({"format":"trainlog-sync-orchestrator-config","version":1,"enabled":True,"mode":"directory-test-double","capabilities":["generation-manifest-v1","generation-ack-v1","causal-delete-v1","mobile-history-v4","execution-draft-v1"],"peer_command":["python3",str(FIXTURE)],"timeout_seconds":30})) - result=subprocess.run(["python3",str(ORCHESTRATOR),"--database",str(db),"--state",str(state),"--config",str(config),"--run-id",run,"--request-id",request],capture_output=True,text=True) - self.assertEqual(result.returncode,0,result.stderr) - final=json.loads(state.read_text());self.assertEqual(final["phase"],"completed") - self.assertEqual(final["result"],"completed");self.assertEqual(final["sessions_reconciled"],1) - self.assertTrue(final["domains"]["causal-delete-v1"]);self.assertEqual(final["drafts"][0]["state"],"pending") - - def test_missing_capability_is_explicitly_degraded(self): + 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}): + 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);db=root/"trainlog.db";db.touch();state=Path(str(db)+".sync-run.json") - run="sy_"+str(uuid.uuid4());request="sy_"+str(uuid.uuid4()) - state.write_text(json.dumps({"run_id":run,"request_id":request,"trigger":"web","phase":"requested","progress_revision":1,"result":"running"})) - config=root/"config.json";config.write_text(json.dumps({"format":"trainlog-sync-orchestrator-config","version":1,"enabled":True,"mode":"directory-test-double","capabilities":[],"peer_command":["python3",str(FIXTURE)],"timeout_seconds":30})) - result=subprocess.run(["python3",str(ORCHESTRATOR),"--database",str(db),"--state",str(state),"--config",str(config),"--run-id",run,"--request-id",request]) - self.assertEqual(result.returncode,3);self.assertEqual(json.loads(state.read_text())["phase"],"explicitly_degraded") + 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()= 20 and connection.execute("SELECT 1 FROM sync_causal_state WHERE target_kind='execution_draft' AND target_id=? AND deleted=1", (session_id,)).fetchone(): outcomes.append((session_id, "stale")); continue if connection.execute("SELECT 1 FROM execution_draft_finalizations WHERE session_id=?", (session_id,)).fetchone(): @@ -192,7 +206,8 @@ def import_document(connection, document, own_transaction=True): "SELECT parent_revision_id,payload_json FROM execution_draft_revisions WHERE session_id=? AND revision_id=?", (session_id, revision)).fetchone() if known: - if tuple(known) != (draft["parent_revision_id"], payload): + if (known[0] != draft["parent_revision_id"] or + causal_payload(json.loads(known[1])) != causal): raise LifecycleError("contradictory revision: " + session_id) outcomes.append((session_id, "unchanged" if current and current[0] == revision else "stale")); continue if current and draft["parent_revision_id"] != current[0]: diff --git a/tools/sync_orchestrator.py b/tools/sync_orchestrator.py index 7931501..1477941 100644 --- a/tools/sync_orchestrator.py +++ b/tools/sync_orchestrator.py @@ -13,8 +13,12 @@ import json import os import subprocess import signal +import selectors +import re import sys import tempfile +import time +import ctypes from datetime import datetime, timezone from pathlib import Path @@ -26,6 +30,29 @@ PHASES = {"requested", "waiting_android_publication", "running", "peer_consumed", "completed", "failed", "interrupted", "explicitly_degraded"} +class OrchestratorInterrupted(RuntimeError): + pass + + +_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 + raise OrchestratorInterrupted(f"orchestrator received signal {signum}") + + +def _child_setup() -> None: + """Create an owned group and ask Linux to terminate it if this parent dies.""" + os.setsid() + libc = ctypes.CDLL(None) + if libc.prctl(1, signal.SIGTERM) != 0: # PR_SET_PDEATHSIG + raise OSError("cannot establish parent-death signal") + + def canonical(value: object) -> bytes: return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode() @@ -51,9 +78,10 @@ def atomic_write(path: Path, value: dict) -> None: def load_json(path: Path, limit: int) -> dict: - raw = path.read_bytes() + with path.open("rb") as stream: + raw = stream.read(limit + 1) if len(raw) > limit: raise RuntimeError(f"{path.name} exceeds its bound") - value = json.loads(raw) + value = json.loads(raw.decode("utf-8", "strict")) if not isinstance(value, dict): raise RuntimeError(f"{path.name} must contain an object") return value @@ -78,9 +106,9 @@ def validate_result(value: dict, run_id: str) -> dict: (value["consumer_peer_id"], "peer_"), (value["inbound_generation_id"], "gen_"), (value["outbound_generation_id"], "gen_")): - if not isinstance(identity, str) or not identity.startswith(prefix) or len(identity) > 64: + 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 not isinstance(value["sessions_reconciled"], int) or value["sessions_reconciled"] < 0: + 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()): raise RuntimeError("invalid domain coverage") @@ -89,26 +117,65 @@ def validate_result(value: dict, run_id: str) -> dict: return value +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 + deadline = time.monotonic() + timeout + 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)): + chunk = os.read(key.fileobj.fileno(), 4096) + 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") + 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") + 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: + 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 + child.wait() + if line_buffer.strip(): raise RuntimeError("unterminated worker frame") + if child.returncode != 0: + 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"]) + + def run(args: argparse.Namespace) -> int: + global _owned_process_group state_path, config_path = Path(args.state), Path(args.config) state = load_json(state_path, MAX_REPORT) 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", "capabilities", "peer_command", "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: raise RuntimeError("full-generation synchronization is disabled") - required_caps = {"generation-manifest-v1", "generation-ack-v1", "causal-delete-v1", "mobile-history-v4", "execution-draft-v1"} - caps = config["capabilities"] - if not isinstance(caps, list) or not required_caps.issubset(set(caps)): - update(state_path, state, "explicitly_degraded", result="incompatible", - missing_capabilities=sorted(required_caps - set(caps if isinstance(caps, list) else [])), - diagnostic="Peer lacks mandatory full-generation capabilities", finished_at=datetime.now(timezone.utc).isoformat()) - return 3 - command = config["peer_command"] - if not isinstance(command, list) or not command or len(command) > 16 or not all(isinstance(part, str) and 0 < len(part) <= 4096 for part in command): - raise RuntimeError("invalid trusted peer command") + 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") timeout = config["timeout_seconds"] if not isinstance(timeout, int) or timeout < 1 or timeout > 900: raise RuntimeError("invalid worker timeout") @@ -121,38 +188,27 @@ def run(args: argparse.Namespace) -> int: except BlockingIOError: 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, "waiting_android_publication", result="running", diagnostic="") - update(state_path, state, "running") + 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)] child = subprocess.Popen(command, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=environment, - start_new_session=True) - try: - stdout, stderr = child.communicate(timeout=timeout) - except subprocess.TimeoutExpired: - os.killpg(child.pid, signal.SIGTERM) - try: stdout, stderr = child.communicate(timeout=5) - except subprocess.TimeoutExpired: - os.killpg(child.pid, signal.SIGKILL); stdout, stderr = child.communicate() - raise RuntimeError("peer worker timed out") - if len(stdout) > MAX_REPORT or len(stderr) > MAX_REPORT: - raise RuntimeError("peer worker output exceeds 64 KiB") - if child.returncode != 0: - detail = stderr.decode("utf-8", "replace").strip()[:MAX_DIAGNOSTIC] - raise RuntimeError(f"peer worker exited {child.returncode}: {detail}") - result = validate_result(json.loads(stdout), args.run_id) - update(state_path, state, "local_import_committed", - inbound_generation_id=result["inbound_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"]) - update(state_path, state, "published", outbound_generation_id=result["outbound_generation_id"]) - update(state_path, state, "waiting_acknowledgement") - update(state_path, state, "peer_consumed", manifest_sha256=result["manifest_sha256"]) + preexec_fn=_child_setup) + _owned_process_group = child.pid + 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() + 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()) return 0 @@ -166,11 +222,14 @@ def main() -> int: 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: path = Path(args.state); state = load_json(path, MAX_REPORT) - update(path, state, "failed", result="failed", diagnostic=str(error)[:MAX_DIAGNOSTIC], + 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 return 2 diff --git a/tools/sync_peer_worker.py b/tools/sync_peer_worker.py new file mode 100644 index 0000000..b19fa29 --- /dev/null +++ b/tools/sync_peer_worker.py @@ -0,0 +1,86 @@ +#!/usr/bin/env python3 +"""Production full-generation peer conversation over a transport directory. + +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 +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"} + +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") + 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 wait_file(path:Path,deadline:float)->None: + while not path.is_file(): + if time.monotonic()>=deadline: raise RuntimeError(f"timeout waiting for {path.name}") + time.sleep(.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"])): + 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) + args=parser.parse_args();deadline=time.monotonic()+args.timeout + 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) + 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";wait_file(inbound_ref,deadline) + inbound=json.loads(bounded(inbound_ref));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 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) + 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) + 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";wait_file(android_ack,deadline) + android_ack_value=json.loads(bounded(android_ack)) + 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 + +if __name__=="__main__": raise SystemExit(main()) diff --git a/tui/src/web_server.c b/tui/src/web_server.c index c0dac14..35e0951 100644 --- a/tui/src/web_server.c +++ b/tui/src/web_server.c @@ -244,11 +244,21 @@ static bool write_initial_sync_state(const TrainlogWebContext *context, static bool launch_sync_worker(TrainlogWebContext *context, const char *config, const char *run_id, const char *request_id) { + char orchestrator[PATH_MAX]; + const char *tools = getenv("TRAINLOG_SYNC_TOOLS_DIR"); + int path_length; + if (tools == NULL) tools = TRAINLOG_TOOLS_DIR; + /* CONTRACT: deployment may relocate the fixed shipped helper directory, + * but an HTTP request can never select this executable or its arguments. */ + path_length = snprintf(orchestrator, sizeof(orchestrator), + "%s/sync_orchestrator.py", tools); + if (tools[0] != '/' || path_length < 0 || + (size_t)path_length >= sizeof(orchestrator)) return false; pid_t child = fork(); if (child < 0) return false; if (child == 0) { (void)setpgid(0, 0); - execlp("python3", "python3", TRAINLOG_TOOLS_DIR "/sync_orchestrator.py", + execlp("python3", "python3", orchestrator, "--database", context->database_path, "--state", context->state_path, "--config", config, "--run-id", run_id, "--request-id", request_id, (char *)NULL);