test: stabilize Windows concurrency checks

Author: guima-why

tests/mcp/test_oauth.py

@@ -1363,17 +1363,16 @@ def wait_for_barrier(name: str) -> None:
                     time.sleep(0.01)
 
             class BarrierStorage(MCPSecretStorage):
-                _blob_barrier_fired = False
+                _blob_reads = 0
 
                 def get_secret(self, key: str) -> str | None:
                     value = super().get_secret(key)
                     # 所有 OAuth 字段现在合并进单个 blob 条目;两个进程都读到过期旧状态后,
                     # 用一次性栅栏让它们同时进入刷新竞争,验证粗粒度 CAS 锁只放行一次网络刷新。
-                    if not self._blob_barrier_fired and key == oauth_module.oauth_storage_key(
-                        config, scope=MCPConfigScope.USER
-                    ):
-                        self._blob_barrier_fired = True
-                        wait_for_barrier("oauth-blob-read")
+                    if key == oauth_module.oauth_storage_key(config, scope=MCPConfigScope.USER):
+                        self._blob_reads += 1
+                        if self._blob_reads == 4:
+                            wait_for_barrier("oauth-blob-read")
                     return value
 
             oauth_module.discover_oauth_metadata = lambda _config: oauth_module.OAuthMetadata(

tests/skill_bridge/test_alicloud_ros_agent_bridge.py

@@ -2298,15 +2298,19 @@ def test_manager_idle_countdown_starts_after_sse_worker_exits(monkeypatch, tmp_p
     monkeypatch.setenv(bridge.STATE_DIR_ENV, str(tmp_path / "state"))
     workspace = tmp_path / "workspace"
     workspace.mkdir()
-    fake_cli = _write_fake_aliyun(
-        tmp_path,
+    # Invoke the current interpreter as the fake CLI and let it execute the
+    # positional ``ros`` script from the worker cwd. This avoids depending on
+    # Windows batch-file launch behavior in a manager lifecycle test.
+    fake_cli = Path(sys.executable)
+    (workspace / "ros").write_text(
         "import json, time\n"
         + "time.sleep(0.6)\n"
         + "event = {'result': {'statusUpdate': {'taskId': 'task-1', 'contextId': 'session-1', "
         + "'status': {'state': 'TASK_STATE_INPUT_REQUIRED', 'message': {'role': 'ROLE_AGENT', "
         + "'parts': [{'text': 'done'}]}}, 'metadata': {'iac_code': {'assistantFinal': "
         + "{'complete': True}}, 'iacCodeSessionId': 'iac-1'}}}}\n"
         + "print(json.dumps({'data': event}), flush=True)\n",
+        encoding="utf-8",
     )
 
     # Leave enough scheduling headroom for a loaded Windows xdist runner; this