ci: fix pre dryrun fail

.github/workflows/sync.yml

@@ -128,21 +128,24 @@ jobs:
           HF_TOKEN: ${{ secrets.HF_TOKEN }}
           MODELSCOPE_TOKEN: ${{ secrets.MODELSCOPE_TOKEN }}
         run: |
-          ARGS="--config config/sync_config.yaml --state-dir .sync_state/"
+          ARGS=(
+              "--config" "config/sync_config.yaml"
+              "--state-dir" ".sync_state/"
+          )
 
           if [ "${{ inputs.dry_run }}" = "true" ]; then
-            ARGS="$ARGS --dry-run true"
+            ARGS+=("--dry-run" "true")
           fi
 
           if [ -n "${{ inputs.direction }}" ] && [ "${{ inputs.direction }}" != "config" ]; then
-            ARGS="$ARGS --direction ${{ inputs.direction }}"
+            ARGS+=("--direction" "${{ inputs.direction }}")
           fi
 
           if [ -n "${{ inputs.sync_target }}" ]; then
-            ARGS="$ARGS --target ${{ inputs.sync_target }}"
+            ARGS+=("--target" "${{ inputs.sync_target }}")
           fi
 
-          python -m src.sync_engine $ARGS
+          python -m src.sync_engine "${ARGS[@]}"
 
       - name: Save sync state
         uses: actions/upload-artifact@v4

config/sync_config.yaml

@@ -16,7 +16,7 @@ global:
 models:
   - name: "qwen2.5-7b-instruct"
     hf_repo_id: "Qwen/Qwen2.5-7B-Instruct"
-    ms_repo_id: "Qwen/Qwen2.5-7B-Instruct"
+    ms_repo_id: "dongjiang1989/Qwen2.5-7B-Instruct"
     direction: "hf_to_ms"
     include_patterns:
       - "*.safetensors"

src/adapters/modelscope_adapter.py

@@ -211,6 +211,15 @@ def upload_file(
                 repo_type="dataset" if resource_type == "dataset" else "model",
             )
         except Exception as e:
+            error_msg = str(e).lower()
+            # Detect permission errors and provide clear message
+            perm_keywords = ("does not exist", "403", "401", "forbidden")
+            if any(kw in error_msg for kw in perm_keywords):
+                raise PermissionError(
+                    f"Cannot upload to {repo_id}: you don't have write access. "
+                    f"Make sure you own this repo or use your own namespace "
+                    f"(e.g., 'your-username/model-name' instead of 'Qwen/model-name')."
+                ) from e
             logger.warning("[MS] SDK upload failed: %s, trying HF fallback", e)
             self._upload_via_hf(repo_id, local_path, remote_path, resource_type)
 
@@ -225,32 +234,59 @@ def _upload_via_hf(
 
         api = HfApi(endpoint="https://modelscope.cn", token=self._token)
         repo_type = "dataset" if resource_type == "dataset" else "model"
-        with open(local_path, "rb") as f:
-            api.upload_file(
-                path_or_fileobj=f,
-                path_in_repo=remote_path,
-                repo_id=repo_id,
-                repo_type=repo_type,
-            )
+        try:
+            with open(local_path, "rb") as f:
+                api.upload_file(
+                    path_or_fileobj=f,
+                    path_in_repo=remote_path,
+                    repo_id=repo_id,
+                    repo_type=repo_type,
+                )
+        except Exception as e:
+            error_msg = str(e).lower()
+            # Detect permission errors and provide clear message
+            perm_keywords = ("does not exist", "403", "401", "forbidden")
+            if any(kw in error_msg for kw in perm_keywords):
+                raise PermissionError(
+                    f"Cannot upload to {repo_id}: you don't have write access. "
+                    f"Make sure you own this repo or use your own namespace "
+                    f"(e.g., 'your-username/model-name' instead of 'Qwen/model-name')."
+                ) from e
+            raise
 
     def create_repo_if_needed(
         self,
         repo_id: str,
         resource_type: Literal["model", "dataset"],
     ) -> None:
         try:
-            self.get_repo_snapshot(repo_id, resource_type)
-            logger.info("[MS] Repo %s already exists", repo_id)
-        except Exception:
-            logger.info("[MS] Creating repo %s (%s)", repo_id, resource_type)
             if self._api:
-                self._api.create_repo(repo_id)
+                exists = self._api.repo_exists(repo_id)
             else:
                 from huggingface_hub import HfApi
 
                 api = HfApi(endpoint="https://modelscope.cn", token=self._token)
+                exists = api.repo_exists(
+                    repo_id=repo_id,
+                    repo_type="dataset" if resource_type == "dataset" else "model",
+                )
+
+            if exists:
+                logger.info("[MS] Repo %s already exists", repo_id)
+                return
+
+            logger.info("[MS] Creating repo %s (%s)", repo_id, resource_type)
+            if self._api:
+                self._api.create_repo(
+                    repo_id=repo_id,
+                    repo_type="dataset" if resource_type == "dataset" else "model",
+                )
+            else:
                 api.create_repo(
                     repo_id=repo_id,
                     repo_type="dataset" if resource_type == "dataset" else "model",
                     exist_ok=True,
                 )
+        except Exception as e:
+            logger.warning("[MS] create_repo_if_needed failed for %s: %s", repo_id, e)
+            # Non-fatal: sync will fail naturally if repo doesn't exist

src/sync_engine.py

@@ -133,7 +133,12 @@ def sync_item(self, item: SyncItem) -> SyncResult:
                 self.hf.create_repo_if_needed(item.hf_repo_id, item.resource_type)
 
             # Fetch snapshots
-            hf_snapshot = self.hf.get_repo_snapshot(item.hf_repo_id, item.resource_type)
+            hf_snapshot = None
+            try:
+                hf_snapshot = self.hf.get_repo_snapshot(item.hf_repo_id, item.resource_type)
+            except Exception as e:
+                logger.warning("[HF] Could not fetch snapshot for %s: %s", item.hf_repo_id, e)
+
             ms_snapshot = None
             try:
                 ms_snapshot = self.ms.get_repo_snapshot(item.ms_repo_id, item.resource_type)
@@ -150,6 +155,12 @@ def sync_item(self, item: SyncItem) -> SyncResult:
             )
 
             if item.direction == SyncDirection.BIDIRECTIONAL:
+                if hf_snapshot is None or ms_snapshot is None:
+                    raise RuntimeError(
+                        f"Cannot sync BIDIRECTIONAL: failed to fetch snapshot — "
+                        f"HF ({item.hf_repo_id})={'OK' if hf_snapshot else 'FAILED'}, "
+                        f"MS ({item.ms_repo_id})={'OK' if ms_snapshot else 'FAILED'}"
+                    )
                 bd = BidirectionalChangeDetector(**detector_kwargs)
                 hf_state = self.states.get(
                     SyncState.make_key("hf", item.resource_type, item.hf_repo_id)
@@ -165,6 +176,11 @@ def sync_item(self, item: SyncItem) -> SyncResult:
                 )
                 all_actions = hf_to_ms_actions + ms_to_hf_actions
             elif item.direction == SyncDirection.HF_TO_MS:
+                if hf_snapshot is None:
+                    raise RuntimeError(
+                        f"Cannot sync HF_TO_MS: failed to fetch HuggingFace "
+                        f"snapshot for {item.hf_repo_id}"
+                    )
                 detector = ChangeDetector(**detector_kwargs)
                 state = self.states.get(
                     SyncState.make_key("hf", item.resource_type, item.hf_repo_id)
@@ -233,8 +249,10 @@ def sync_item(self, item: SyncItem) -> SyncResult:
                     else (SyncStatus.PARTIAL if success else SyncStatus.FAILED)
                 )
 
-            # Update sync state
-            self._update_state(item, hf_snapshot, ms_snapshot, result.files_synced)
+            # Update sync state even on partial success, so successfully
+            # transferred files are recorded and won't be re-synced next run
+            if result.files_synced:
+                self._update_state(item, hf_snapshot, ms_snapshot, result.files_synced)
 
         except Exception as e:
             logger.error("Failed to sync %s: %s", item.name, e, exc_info=True)
@@ -250,7 +268,11 @@ def _execute_transfers(
         item: SyncItem,
         max_parallel: int,
     ) -> tuple[list[str], list[str], int]:
-        """Execute file transfers with concurrency control.
+        """Execute file transfers with disk-space awareness.
+
+        Large files (>100MB) are transferred sequentially to avoid filling
+        up disk (common on CI runners with limited space).
+        Small files are transferred in parallel for speed.
 
         Returns (success_files, failed_files, total_bytes).
         """
@@ -259,20 +281,59 @@ def _execute_transfers(
         total_bytes = 0
         temp_dir = create_temp_dir()
 
+        large_file_threshold = 100 * 1024 * 1024  # 100 MB
+
         try:
             # Sort: small files first for faster feedback
             sorted_actions = sorted(actions, key=lambda a: a.size)
 
-            with ThreadPoolExecutor(max_workers=max_parallel) as pool:
-                futures = {}
-                for action in sorted_actions:
-                    future = pool.submit(self._transfer_file, action, item, temp_dir)
-                    futures[future] = action
+            large_actions = [a for a in sorted_actions if a.size > large_file_threshold]
+            small_actions = [a for a in sorted_actions if a.size <= large_file_threshold]
 
-                for future in as_completed(futures):
-                    action = futures[future]
+            # Transfer small files in parallel
+            if small_actions:
+                logger.info(
+                    "Transferring %d small files in parallel (max_workers=%d)",
+                    len(small_actions),
+                    max_parallel,
+                )
+                with ThreadPoolExecutor(max_workers=max_parallel) as pool:
+                    futures = {}
+                    for action in small_actions:
+                        future = pool.submit(self._transfer_file, action, item, temp_dir)
+                        futures[future] = action
+
+                    for future in as_completed(futures):
+                        action = futures[future]
+                        try:
+                            transferred_bytes = future.result()
+                            success.append(action.file_path)
+                            total_bytes += transferred_bytes
+                            logger.info(
+                                "  ✓ %s (%s)",
+                                action.file_path,
+                                format_bytes(transferred_bytes),
+                            )
+                        except Exception as e:
+                            failed.append(action.file_path)
+                            logger.error("  ✗ %s: %s", action.file_path, e)
+
+            # Transfer large files sequentially to save disk space
+            if large_actions:
+                logger.info(
+                    "Transferring %d large files sequentially to conserve disk space",
+                    len(large_actions),
+                )
+                for i, action in enumerate(large_actions, 1):
+                    logger.info(
+                        "  [%d/%d] %s (%s)",
+                        i,
+                        len(large_actions),
+                        action.file_path,
+                        format_bytes(action.size),
+                    )
                     try:
-                        transferred_bytes = future.result()
+                        transferred_bytes = self._transfer_file(action, item, temp_dir)
                         success.append(action.file_path)
                         total_bytes += transferred_bytes
                         logger.info(
@@ -350,10 +411,11 @@ def _update_state(
         # Update HF state
         hf_key = SyncState.make_key("hf", item.resource_type, item.hf_repo_id)
         hf_state = self.states.get(hf_key, SyncState(repo_key=hf_key))
-        hf_state.last_synced_commit = hf_snapshot.last_commit_hash
+        if hf_snapshot:
+            hf_state.last_synced_commit = hf_snapshot.last_commit_hash
         hf_state.last_synced_at = now
         for fp in synced_files:
-            hf_file = hf_file_map.get(fp)
+            hf_file = hf_file_map.get(fp) or ms_file_map.get(fp)
             if hf_file and hf_file.sha256:
                 hf_state.synced_files[fp] = hf_file.sha256
         self.states[hf_key] = hf_state
@@ -365,7 +427,7 @@ def _update_state(
             ms_state.last_synced_commit = ms_snapshot.last_commit_hash
         ms_state.last_synced_at = now
         for fp in synced_files:
-            ms_file = ms_file_map.get(fp)
+            ms_file = ms_file_map.get(fp) or hf_file_map.get(fp)
             if ms_file and ms_file.sha256:
                 ms_state.synced_files[fp] = ms_file.sha256
         self.states[ms_key] = ms_state
@@ -516,8 +578,10 @@ def main() -> None:
 
     print_summary(results)
 
-    # Exit with error code if any failures
-    if any(r.status == SyncStatus.FAILED for r in results):
+    # Exit with error code if any item failed.
+    # Downstream CI steps (report, issue creation) use `if: always()`.
+    any_failed = any(r.status == SyncStatus.FAILED for r in results)
+    if any_failed:
         sys.exit(1)
 
 

tests/test_sync_engine_extended.py

@@ -310,6 +310,65 @@ def test_snapshot_exception_caught(self, tmp_path):
         results = engine.sync_all()
         assert results[0].status == SyncStatus.FAILED
 
+    def test_both_snapshots_fail_bidirectional(self, tmp_path):
+        """Both snapshots fail in bidirectional mode — item should be FAILED."""
+        hf = MockAdapter("hf")
+        hf._snapshot_fail = True
+        ms = MockAdapter("ms")
+        ms._snapshot_fail = True
+
+        config = make_config(direction="bidirectional")
+        engine = SyncEngine(
+            config=config,
+            hf_adapter=hf,
+            ms_adapter=ms,
+            state_dir=tmp_path,
+        )
+
+        results = engine.sync_all()
+        assert results[0].status == SyncStatus.FAILED
+        err = results[0].error_message.lower()
+        assert "failed" in err
+
+    def test_one_snapshot_fails_bidirectional(self, tmp_path):
+        """Only one snapshot fails in bidirectional mode — should still be FAILED."""
+        hf = MockAdapter("hf")
+        hf._snapshot_fail = True
+        ms = MockAdapter("ms")
+        ms._files = {"model.bin": b"data"}
+
+        config = make_config(direction="bidirectional")
+        engine = SyncEngine(
+            config=config,
+            hf_adapter=hf,
+            ms_adapter=ms,
+            state_dir=tmp_path,
+        )
+
+        results = engine.sync_all()
+        assert results[0].status == SyncStatus.FAILED
+        err = results[0].error_message.lower()
+        assert "hf" in err and "failed" in err
+
+    def test_hf_snapshot_fails_ms_to_hf_ok(self, tmp_path):
+        """HF snapshot fails but MS_TO_HF only needs MS snapshot — should succeed."""
+        hf = MockAdapter("hf")
+        hf._snapshot_fail = True
+        hf._files = {}
+        ms = MockAdapter("ms")
+        ms._files = {"model.bin": b"data"}
+
+        config = make_config(direction="ms_to_hf")
+        engine = SyncEngine(
+            config=config,
+            hf_adapter=hf,
+            ms_adapter=ms,
+            state_dir=tmp_path,
+        )
+
+        results = engine.sync_all()
+        assert results[0].status == SyncStatus.SUCCESS
+
 
 # ── Results JSON writing ────────────────────────────────────────────