fix(sync): arbitrate shared mtp transport

tests/test_sync_error_classification.py

@@ -129,18 +129,21 @@ class SyncErrorClassificationTest(unittest.TestCase):
 
     def test_transport_timeout_has_a_stable_code_and_bounded_diagnostic(self):
         timeout = subprocess.TimeoutExpired(["adapter", "push"], 30)
-        with mock.patch.object(sync_peer_worker.subprocess, "run", side_effect=timeout):
+        with tempfile.TemporaryDirectory() as directory:
-            with self.assertRaisesRegex(
+            transport = Path(directory) / "transport"
-                RuntimeError,
+            transport.mkdir()
-                r"^transport_timeout: MTP push did not finish within 30 seconds$",
+            with mock.patch.object(sync_peer_worker.subprocess, "run", side_effect=timeout):
-            ) as caught:
+                with self.assertRaisesRegex(
-                sync_peer_worker.run_adapter(
+                    RuntimeError,
-                    Path("/adapter"),
+                    r"^transport_timeout: MTP push did not finish within 30 seconds$",
-                    "push",
+                ) as caught:
-                    "peer_00000000-0000-4000-8000-000000000001",
+                    sync_peer_worker.run_adapter(
-                    Path("/transport"),
+                        Path("/adapter"),
-                    sync_peer_worker.time.monotonic() + 60,
+                        "push",
-                )
+                        "peer_00000000-0000-4000-8000-000000000001",
+                        transport,
+                        sync_peer_worker.time.monotonic() + 60,
+                    )
         self.assertEqual(sync_orchestrator.failure_code(caught.exception), "transport_timeout")
 
     def test_known_operational_failures_remain_distinct(self):

tests/test_sync_peer_worker.py

@@ -17,6 +17,14 @@ SPEC.loader.exec_module(worker)
 
 
 class SyncPeerWorkerTest(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory(prefix="trainlog-peer-worker-test-")
+        self.transport = Path(self.temporary.name) / "transport"
+        self.transport.mkdir()
+
+    def tearDown(self):
+        self.temporary.cleanup()
+
     def test_drive_adapter_uses_remaining_conversation_deadline(self):
         completed = subprocess.CompletedProcess([], 0, b"", b"")
         with mock.patch.object(worker.subprocess, "run", return_value=completed) as run:
@@ -24,7 +32,7 @@ class SyncPeerWorkerTest(unittest.TestCase):
                 Path("/fixed/adapter.py"),
                 "pull",
                 "remote:Trainlog/Sync-Test/run",
-                Path("/private/transport"),
+                self.transport,
                 time.monotonic() + 180,
             )
         timeout = run.call_args.kwargs["timeout"]
@@ -36,10 +44,25 @@ class SyncPeerWorkerTest(unittest.TestCase):
         with mock.patch.object(worker.subprocess, "run", return_value=completed) as run:
             worker.run_adapter(
                 Path("/fixed/adapter"), "pull", "peer_fixed",
-                Path("/private/transport"), 100.0, clock=lambda: 50.0,
+                self.transport, 100.0, clock=lambda: 50.0,
             )
         self.assertEqual(worker.MTP_OPERATION_TIMEOUT_SECONDS, run.call_args.kwargs["timeout"])
 
+    def test_mtp_adapter_uses_and_releases_shared_transport_lock(self):
+        completed = subprocess.CompletedProcess([], 0, b"", b"")
+        real_flock = worker.fcntl.flock
+        with mock.patch.object(worker.subprocess, "run", return_value=completed), mock.patch.object(
+            worker.fcntl, "flock", wraps=real_flock
+        ) as flock:
+            worker.run_adapter(
+                Path("/fixed/adapter"), "pull", "peer_fixed",
+                self.transport, 100.0, clock=lambda: 50.0,
+            )
+        self.assertTrue((self.transport.parent / "mtp.lock").is_file())
+        operations = [call.args[1] for call in flock.call_args_list]
+        self.assertIn(worker.fcntl.LOCK_EX | worker.fcntl.LOCK_NB, operations)
+        self.assertEqual(worker.fcntl.LOCK_UN, operations[-1])
+
     def test_mtp_pull_pump_is_throttled(self):
         now = [0.0]
         pulls = []
@@ -55,7 +78,7 @@ class SyncPeerWorkerTest(unittest.TestCase):
         with self.assertRaises(worker.ConversationDeadlineExpired):
             worker.run_adapter(
                 Path("/fixed/adapter"), "pull", "peer_fixed",
-                Path("/private/transport"), 10.0, clock=lambda: 10.0,
+                self.transport, 10.0, clock=lambda: 10.0,
             )
 
     def test_deadline_limited_pull_becomes_protocol_timeout(self):
@@ -65,7 +88,7 @@ class SyncPeerWorkerTest(unittest.TestCase):
             with self.assertRaises(worker.ConversationDeadlineExpired):
                 worker.run_adapter(
                     Path("/fixed/adapter"), "pull", "peer_fixed",
-                    Path("/private/transport"), 10.5, clock=lambda: 10.0,
+                    self.transport, 10.5, clock=lambda: 10.0,
                 )
 
     def test_blocked_full_budget_pull_is_transport_timeout(self):
@@ -75,7 +98,7 @@ class SyncPeerWorkerTest(unittest.TestCase):
             with self.assertRaisesRegex(RuntimeError, "transport_timeout"):
                 worker.run_adapter(
                     Path("/fixed/adapter"), "pull", "peer_fixed",
-                    Path("/private/transport"), 100.0, clock=lambda: 50.0,
+                    self.transport, 100.0, clock=lambda: 50.0,
                 )
 
     def test_delayed_correlated_generation_survives_multiple_polls(self):

tests/test_syncd_routing.py

@@ -126,7 +126,7 @@ class SyncdRoutingTest(unittest.TestCase):
             self.assertEqual(1, len(commands))
             self.assertEqual("sy_old", json.loads(state.read_text())["run_id"])
 
-    def test_daemon_skips_mtp_probe_while_canonical_sync_lock_is_busy(self):
+    def test_daemon_skips_mtp_probe_while_transport_lock_is_busy(self):
         with tempfile.TemporaryDirectory() as directory:
             root = Path(directory)
             transport = root / "transport"

tools/sync_peer_worker.py

@@ -9,6 +9,7 @@ always produced and consumed by the existing generation services.
 from __future__ import annotations
 
 import argparse
+import fcntl
 import hashlib
 import json
 import os
@@ -123,28 +124,46 @@ def run_adapter(
     allow_missing_peer: bool = False,
     clock=time.monotonic,
 ) -> bool:
-    remaining = deadline - clock()
+    lock_path = root.parent / "mtp.lock"
-    if remaining <= 0:
+    lock_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
-        raise ConversationDeadlineExpired()
+    lock_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600)
-    timeout = min(remaining, MTP_OPERATION_TIMEOUT_SECONDS)
-    deadline_limited = remaining < MTP_OPERATION_TIMEOUT_SECONDS
     try:
-        result = subprocess.run(
+        while True:
-            [str(adapter), operation, peer, str(root)],
+            remaining = deadline - clock()
-            stdin=subprocess.DEVNULL,
+            if remaining <= 0:
-            capture_output=True,
+                raise ConversationDeadlineExpired()
-            timeout=timeout,
+            try:
-            check=False,
+                fcntl.flock(lock_fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
-        )
+                break
-    except subprocess.TimeoutExpired as error:
+            except BlockingIOError:
-        if deadline_limited:
+                time.sleep(min(0.05, remaining))
-            raise ConversationDeadlineExpired() from error
+        remaining = deadline - clock()
-        # CONTRACT: an adapter timeout is an ambiguous transport failure. It
+        if remaining <= 0:
-        # never implies rollback, replay, or permission to retry a mutation.
+            raise ConversationDeadlineExpired()
-        raise RuntimeError(
+        timeout = min(remaining, MTP_OPERATION_TIMEOUT_SECONDS)
-            f"transport_timeout: MTP {operation} did not finish within "
+        deadline_limited = remaining < MTP_OPERATION_TIMEOUT_SECONDS
-            f"{MTP_OPERATION_TIMEOUT_SECONDS:g} seconds"
+        try:
-        ) from error
+            result = subprocess.run(
+                [str(adapter), operation, peer, str(root)],
+                stdin=subprocess.DEVNULL,
+                capture_output=True,
+                timeout=timeout,
+                check=False,
+            )
+        except subprocess.TimeoutExpired as error:
+            if deadline_limited:
+                raise ConversationDeadlineExpired() from error
+            # CONTRACT: an adapter timeout is an ambiguous transport failure. It
+            # never implies rollback, replay, or permission to retry a mutation.
+            raise RuntimeError(
+                f"transport_timeout: MTP {operation} did not finish within "
+                f"{MTP_OPERATION_TIMEOUT_SECONDS:g} seconds"
+            ) from error
+    finally:
+        try:
+            fcntl.flock(lock_fd, fcntl.LOCK_UN)
+        finally:
+            os.close(lock_fd)
     if result.returncode != 0:
         diagnostic = (
             result.stderr.decode("utf-8", "replace")[:1024] or "MTP adapter failed"

tools/trainlog_syncd.py

@@ -170,13 +170,13 @@ def run_full_generation(args: argparse.Namespace, config_path: Path) -> int:
     expected_peer = config["expected_peer_id"]
     adapter = args.mtp_adapter
     # WHY: the daemon must probe MTP to discover Android-origin requests, but
-    # Web/TUI orchestration owns the same physical transport while it holds the
+    # Web/TUI orchestration can use the same libmtp device concurrently. A
-    # canonical sync lock. Probing outside that lock lets libmtp race itself
+    # business sync.lock is too broad here: holding it during discovery makes
-    # and fail with "device is busy". CONTRACT: idle discovery takes the same
+    # a user Web click fail with sync_in_progress. CONTRACT: discovery instead
-    # lock non-blockingly and simply skips this poll when another conversation
+    # shares a dedicated physical-transport lock with every worker MTP adapter
-    # is active. INVARIANT: the lock is released before launching the
+    # call and skips this poll when that transport is busy. INVARIANT: no MTP
-    # orchestrator, which remains the sole owner of business-sync locking.
+    # adapter processes overlap, while business-sync admission stays independent.
-    lock_path = args.database.parent / "sync.lock"
+    lock_path = root.parent / "mtp.lock"
     lock_path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
     lock_fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o600)
     try: